Merge pull request #538 from zendesk/dasch/allow-positional-seed-brokers

Allow passing in the seed brokers in a positional argument

Author: dasch

README.md

@@ -115,13 +115,10 @@ A client must be initialized with at least one Kafka broker, from which the enti
 ```ruby
 require "kafka"
 
-kafka = Kafka.new(
-  # At least one of these nodes must be available:
-  seed_brokers: ["kafka1:9092", "kafka2:9092"],
-
-  # Set an optional client id in order to identify the client to Kafka:
-  client_id: "my-application",
-)
+# The first argument is a list of "seed brokers" that will be queried for the full
+# cluster topology. At least one of these *must* be available. `client_id` is
+# used to identify this client in logs and metrics. It's optional but recommended.
+kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], client_id: "my-application")
 ```
 
 ### Producing Messages to Kafka
@@ -430,10 +427,7 @@ require "kafka"
 
 # Configure the Kafka client with the broker hosts and the Rails
 # logger.
-$kafka = Kafka.new(
-  seed_brokers: ["kafka1:9092", "kafka2:9092"],
-  logger: Rails.logger,
-)
+$kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], logger: Rails.logger)
 
 # Set up an asynchronous producer that delivers its buffered messages
 # every ten seconds:
@@ -473,7 +467,7 @@ Consuming messages from a Kafka topic with ruby-kafka is simple:
 ```ruby
 require "kafka"
 
-kafka = Kafka.new(seed_brokers: ["kafka1:9092", "kafka2:9092"])
+kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
 
 kafka.each_message(topic: "greetings") do |message|
   puts message.offset, message.key, message.value
@@ -496,7 +490,7 @@ Using the API is simple:
 ```ruby
 require "kafka"
 
-kafka = Kafka.new(seed_brokers: ["kafka1:9092", "kafka2:9092"])
+kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
 
 # Consumers with the same group id will form a Consumer Group together.
 consumer = kafka.consumer(group_id: "my-consumer")
@@ -884,10 +878,7 @@ By enabling SSL encryption you can have some confidence that messages can be sen
 In this case you just need to pass a valid CA certificate as a string when configuring your `Kafka` client:
 
 ```ruby
-kafka = Kafka.new(
-  ssl_ca_cert: File.read('my_ca_cert.pem'),
-  # ...
-)
+kafka = Kafka.new(["kafka1:9092"], ssl_ca_cert: File.read('my_ca_cert.pem'))
 ```
 
 Without passing the CA certificate to the client it would be impossible to protect against [man-in-the-middle attacks](https://en.wikipedia.org/wiki/Man-in-the-middle_attack).
@@ -898,10 +889,7 @@ If you want to use the CA certs from your system's default certificate store, yo
 can use:
 
 ```ruby
-kafka = Kafka.new(
-  ssl_ca_certs_from_system: true
-  # ...
-)
+kafka = Kafka.new(["kafka1:9092"], ssl_ca_certs_from_system: true)
 ```
 
 This configures the store to look up CA certificates from the system default certificate store on an as needed basis. The location of the store can usually be determined by: 
@@ -913,6 +901,7 @@ In order to authenticate the client to the cluster, you need to pass in a certif
 
 ```ruby
 kafka = Kafka.new(
+  ["kafka1:9092"],
   ssl_ca_cert: File.read('my_ca_cert.pem'),
   ssl_client_cert: File.read('my_client_cert.pem'),
   ssl_client_cert_key: File.read('my_client_cert_key.pem'),
@@ -935,6 +924,7 @@ In order to authenticate using GSSAPI, set your principal and optionally your ke
 
 ```ruby
 kafka = Kafka.new(
+  ["kafka1:9092"],
   sasl_gssapi_principal: 'kafka/kafka.example.com@EXAMPLE.COM',
   sasl_gssapi_keytab: '/etc/keytabs/kafka.keytab',
   # ...
@@ -946,6 +936,7 @@ In order to authenticate using PLAIN, you must set your username and password wh
 
 ```ruby
 kafka = Kafka.new(
+  ["kafka1:9092"],
   ssl_ca_cert: File.read('/etc/openssl/cert.pem'), # Optional but highly recommended
   sasl_plain_username: 'username',
   sasl_plain_password: 'password'
@@ -960,6 +951,7 @@ Since 0.11 kafka supports [SCRAM](https://kafka.apache.org/documentation.html#se
 
 ```ruby
 kafka = Kafka.new(
+  ["kafka1:9092"],
   sasl_scram_username: 'username',
   sasl_scram_password: 'password',
   sasl_scram_mechanism: 'sha256',

lib/kafka.rb

@@ -243,8 +243,14 @@ class FailedScramAuthentication < SaslScramError
   #
   # @see Client#initialize
   # @return [Client]
-  def self.new(**options)
-    Client.new(**options)
+  def self.new(seed_brokers = nil, **options)
+    # We allow `seed_brokers` to be passed in either as a positional _or_ as a
+    # keyword argument.
+    if seed_brokers.nil?
+      Client.new(**options)
+    else
+      Client.new(seed_brokers: seed_brokers, **options)
+    end
   end
 end
 

lib/kafka/client.rb

@@ -481,7 +481,7 @@ def delete_topic(name, timeout: 30)
     # @note This is an alpha level API and is subject to change.
     #
     # @example Describing the cleanup policy config of a topic
-    #   kafka = Kafka.new(seed_brokers: ["kafka1:9092"])
+    #   kafka = Kafka.new(["kafka1:9092"])
     #   kafka.describe_topic("my-topic", ["cleanup.policy"])
     #   #=> { "cleanup.policy" => "delete" }
     #

lib/kafka/consumer.rb

@@ -20,7 +20,7 @@ module Kafka
   #
   #     require "kafka"
   #
-  #     kafka = Kafka.new(seed_brokers: ["kafka1:9092", "kafka2:9092"])
+  #     kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
   #
   #     # Create a new Consumer instance in the group `my-group`:
   #     consumer = kafka.consumer(group_id: "my-group")

lib/kafka/producer.rb

@@ -14,7 +14,7 @@ module Kafka
   # do it for you, e.g.
   #
   #     # Will instantiate Kafka::Client
-  #     kafka = Kafka.new(...)
+  #     kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
   #
   #     # Will instantiate Kafka::Producer
   #     producer = kafka.producer
@@ -106,12 +106,7 @@ module Kafka
   #     # cluster to auto-create topics.
   #     topic = "random-messages"
   #
-  #     kafka = Kafka.new(
-  #       seed_brokers: brokers,
-  #       client_id: "simple-producer",
-  #       logger: logger,
-  #     )
-  #
+  #     kafka = Kafka.new(brokers, client_id: "simple-producer", logger: logger)
   #     producer = kafka.producer
   #
   #     begin

spec/functional/batch_consumer_spec.rb

@@ -9,7 +9,7 @@
     topic = create_random_topic(num_partitions: 15)
 
     Thread.new do
-      kafka = Kafka.new(seed_brokers: kafka_brokers, client_id: "test")
+      kafka = Kafka.new(kafka_brokers, client_id: "test")
       producer = kafka.producer
 
       messages.each do |i|
@@ -23,7 +23,7 @@
 
     threads = 2.times.map do |thread_id|
       t = Thread.new do
-        kafka = Kafka.new(seed_brokers: kafka_brokers, client_id: "test", logger: logger)
+        kafka = Kafka.new(kafka_brokers, client_id: "test", logger: logger)
         consumer = kafka.consumer(group_id: group_id, offset_retention_time: offset_retention_time)
         consumer.subscribe(topic)
 

spec/functional/client_spec.rb

@@ -10,7 +10,7 @@
 
   example "listing available topics in the cluster" do
     # Use a clean Kafka instance to avoid hitting caches.
-    kafka = Kafka.new(seed_brokers: KAFKA_BROKERS, logger: LOGGER)
+    kafka = Kafka.new(KAFKA_BROKERS, logger: LOGGER)
 
     topics = kafka.topics
 

spec/functional/consumer_group_spec.rb

@@ -6,7 +6,7 @@
     messages = (1..1000).to_a
 
     begin
-      kafka = Kafka.new(seed_brokers: kafka_brokers, client_id: "test")
+      kafka = Kafka.new(kafka_brokers, client_id: "test")
       producer = kafka.producer
 
       messages.each do |i|
@@ -22,7 +22,7 @@
     received_messages = []
 
     consumers = 2.times.map do
-      kafka = Kafka.new(seed_brokers: kafka_brokers, client_id: "test", logger: logger)
+      kafka = Kafka.new(kafka_brokers, client_id: "test", logger: logger)
       consumer = kafka.consumer(group_id: group_id, offset_retention_time: offset_retention_time)
       consumer.subscribe(topic)
       consumer
@@ -60,7 +60,7 @@
     var = ConditionVariable.new
 
     Thread.new do
-      kafka = Kafka.new(seed_brokers: kafka_brokers, client_id: "test")
+      kafka = Kafka.new(kafka_brokers, client_id: "test")
       producer = kafka.producer
 
       messages.each do |i|
@@ -88,7 +88,7 @@
 
     threads = 2.times.map do |thread_id|
       t = Thread.new do
-        kafka = Kafka.new(seed_brokers: kafka_brokers, client_id: "test", logger: logger)
+        kafka = Kafka.new(kafka_brokers, client_id: "test", logger: logger)
         consumer = kafka.consumer(group_id: group_id, offset_retention_time: offset_retention_time)
         consumer.subscribe(topic)
 

spec/fuzz/consumer_group_spec.rb

@@ -10,7 +10,7 @@
   before do
     logger.level = Logger::INFO
 
-    kafka = Kafka.new(seed_brokers: kafka_brokers, logger: logger)
+    kafka = Kafka.new(kafka_brokers, logger: logger)
     producer = kafka.producer(max_buffer_size: num_messages)
 
     messages.each do |i|
@@ -66,7 +66,7 @@
   def start_consumer(group_id, result_queue)
     thread = Thread.new do
       kafka = Kafka.new(
-        seed_brokers: kafka_brokers,
+        kafka_brokers,
         logger: logger,
         socket_timeout: 20,
         connect_timeout: 20,

spec/fuzz/producer_spec.rb

@@ -1,6 +1,6 @@
 describe "Producing a lot of messages with an unreliable cluster", fuzz: true do
   let(:logger) { LOGGER }
-  let(:kafka) { Kafka.new(seed_brokers: kafka_brokers, client_id: "test", logger: logger) }
+  let(:kafka) { Kafka.new(kafka_brokers, client_id: "test", logger: logger) }
   let(:producer) { kafka.producer(max_retries: 20, retry_backoff: 5) }
 
   before do

spec/spec_helper.rb

@@ -69,7 +69,7 @@ def self.included(base)
     base.class_eval do
       let(:logger) { LOGGER }
       let(:kafka_brokers) { KAFKA_BROKERS }
-      let(:kafka) { Kafka.new(seed_brokers: kafka_brokers, client_id: "test", logger: logger) }
+      let(:kafka) { Kafka.new(kafka_brokers, client_id: "test", logger: logger) }
 
       after { kafka.close rescue nil }
     end