Add support for config entries in the topic creation API

Author: tjwp

CHANGELOG.md

@@ -4,6 +4,8 @@ Changes and additions to the library will be listed here.
 
 ## Unreleased
 
+- Add support for config entries in the topic creation API.
+
 ## v0.5.3
 
 - Add support for the topic deletion API (#528).

lib/kafka/client.rb

@@ -457,10 +457,16 @@ def each_message(topic:, start_from_beginning: true, max_wait_time: 5, min_bytes
  # @param replication_factor [Integer] the replication factor of the topic.
  # @param timeout [Integer] a duration of time to wait for the topic to be
  # completely created.
+ # @param config_entries [Hash] topic-level configs to use for the topic.
+ # See https://kafka.apache.org/documentation/#topicconfigs.
  # @raise [Kafka::TopicAlreadyExists] if the topic already exists.
  # @return [nil]
- def create_topic(name, num_partitions: 1, replication_factor: 1, timeout: 30)
- @cluster.create_topic(name, num_partitions: num_partitions, replication_factor: replication_factor, timeout: timeout)
+ def create_topic(name, num_partitions: 1, replication_factor: 1, timeout: 30, config_entries: {})
+ @cluster.create_topic(name,
+ num_partitions: num_partitions,
+ replication_factor: replication_factor,
+ timeout: timeout,
+ config_entries: config_entries)
  end
 
  # Delete a topic in the cluster.

lib/kafka/cluster.rb

@@ -156,12 +156,13 @@ def partitions_for(topic)
  raise
  end
 
- def create_topic(name, num_partitions:, replication_factor:, timeout:)
+ def create_topic(name, num_partitions:, replication_factor:, timeout:, config_entries:)
  options = {
  topics: {
  name => {
  num_partitions: num_partitions,
  replication_factor: replication_factor,
+ config_entries: config_entries,
  }
  },
  timeout: timeout,

lib/kafka/protocol/create_topics_request.rb

@@ -28,7 +28,10 @@ def encode(encoder)
  encoder.write_array([])
 
  # Config entries. We don't care.
- encoder.write_array([])
+ encoder.write_array(config.fetch(:config_entries)) do |config_name, config_value|
+ encoder.write_string(config_name)
+ encoder.write_string(config_value)
+ end
  end
 
  # Timeout is in ms.

spec/functional/topic_management_spec.rb

@@ -10,6 +10,20 @@
  expect(partitions).to eq 3
  end
 
+ example "creating a topic with config entries" do
+ unless kafka.supports_api?(Kafka::Protocol::DESCRIBE_CONFIGS_API)
+ skip("This Kafka version not support ")
+ end
+
+ topic = generate_topic_name
+ expect(kafka.topics).not_to include(topic)
+
+ kafka.create_topic(topic, num_partitions: 3, config_entries: { 'cleanup.policy' => 'compact' })
+
+ configs = kafka.describe_topic(topic, %w(cleanup.policy))
+ expect(configs['cleanup.policy']).to eq('compact')
+ end
+
  example "deleting topics" do
  topic = generate_topic_name
 

spec/spec_helper.rb

@@ -59,8 +59,11 @@ def create_random_topic(*args)
  topic
  end
 
- def create_topic(name, num_partitions: 1, num_replicas: 1)
- kafka.create_topic(name, num_partitions: num_partitions, replication_factor: num_replicas)
+ def create_topic(name, num_partitions: 1, num_replicas: 1, config_entries: {})
+ kafka.create_topic(name,
+ num_partitions: num_partitions,
+ replication_factor: num_replicas,
+ config_entries: config_entries)
  end
 end