Clean up topic create configs

Author: dasch

lib/kafka/client.rb

@@ -451,22 +451,32 @@ def each_message(topic:, start_from_beginning: true, max_wait_time: 5, min_bytes
 
  # Creates a topic in the cluster.
  #
+ # @example Creating a topic with log compaction
+ # # Enable log compaction:
+ # config = { "cleanup.policy" => "compact" }
+ #
+ # # Create the topic:
+ # kafka.create_topic("dns-mappings", config: config)
+ #
  # @param name [String] the name of the topic.
  # @param num_partitions [Integer] the number of partitions that should be created
  # in the topic.
  # @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.
+ # @param config [Hash] topic configuration entries. See
+ # [the Kafka documentation](https://kafka.apache.org/documentation/#topicconfigs)
+ # for more information.
  # @raise [Kafka::TopicAlreadyExists] if the topic already exists.
  # @return [nil]
- 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)
+ def create_topic(name, num_partitions: 1, replication_factor: 1, timeout: 30, config: {})
+ @cluster.create_topic(
+ name,
+ num_partitions: num_partitions,
+ replication_factor: replication_factor,
+ timeout: timeout,
+ config: config,
+ )
  end
 
  # Delete a topic in the cluster.

lib/kafka/cluster.rb

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

lib/kafka/protocol/create_topics_request.rb

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

spec/functional/topic_management_spec.rb

@@ -18,10 +18,10 @@
  topic = generate_topic_name
  expect(kafka.topics).not_to include(topic)
 
- kafka.create_topic(topic, num_partitions: 3, config_entries: { 'cleanup.policy' => 'compact' })
+ kafka.create_topic(topic, num_partitions: 3, config: { "cleanup.policy" => "compact" })
 
- configs = kafka.describe_topic(topic, %w(cleanup.policy))
- expect(configs['cleanup.policy']).to eq('compact')
+ configs = kafka.describe_topic(topic, ["cleanup.policy"])
+ expect(configs.fetch("cleanup.policy")).to eq("compact")
  end
 
  example "deleting topics" do

spec/spec_helper.rb

@@ -59,11 +59,8 @@ def create_random_topic(*args)
  topic
  end
 
- 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)
+ def create_topic(name, num_partitions: 1, num_replicas: 1)
+ kafka.create_topic(name, num_partitions: num_partitions, replication_factor: num_replicas)
  end
 end