add :uris as a valid option for ConsumerGroup
MODIFY
README.md
README.md
@@ -25,6 +25,7 @@ KakfaEx supports the following Kafka features:
* Message Compression with Snappy and gzip
* Offset Management (fetch / commit / autocommit)
* Consumer Groups
+* Topics Management (create / delete)
See [Kafka Protocol Documentation](http://kafka.apache.org/protocol.html) and
[A Guide to the Kafka Protocol](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol)

MODIFY
lib/kafka_ex/consumer_group.ex
lib/kafka_ex/consumer_group.ex
@@ -94,6 +94,7 @@ defmodule KafkaEx.ConsumerGroup do
* `:max_restarts`, `:max_seconds` - Supervisor restart policy parameters
* `:partition_assignment_callback` - See
`t:KafkaEx.ConsumerGroup.PartitionAssignment.callback/0`
+ * `:uris` - See `KafkaEx.create_worker/2`
Note `:session_timeout` is registered with the broker and determines how long
before the broker will de-register a consumer from which it has not heard a
@@ -111,6 +112,7 @@ defmodule KafkaEx.ConsumerGroup do
| {:name, Supervisor.name()}
| {:max_restarts, non_neg_integer}
| {:max_seconds, non_neg_integer}
+ | {:uris, KafkaEx.uri()}
@type options :: [option]

MODIFY
lib/kafka_ex/server_0_p_10_and_later.ex
lib/kafka_ex/server_0_p_10_and_later.ex
@@ -99,7 +99,7 @@ defmodule KafkaEx.Server0P10AndLater do
%KafkaEx.Protocol.ApiVersions.Response{
api_versions: api_versions,
error_code: error_code
- }, state} = kafka_api_versions(%State{brokers: brokers})
+ }, state} = kafka_server_api_versions(%State{brokers: brokers})
if error_code == :no_response do
sleep_for_reconnect()
raise "Brokers sockets are closed"

MODIFY
test/integration/server0_p_10_and_later_test.exs
test/integration/server0_p_10_and_later_test.exs
@@ -20,7 +20,7 @@ defmodule KafkaEx.Server0P10P1AndLater.Test do
resp = create_topic(name, config)
assert {:topic_already_exists, name} == parse_create_topic_resp(resp)
- Enum.member?(existing_topics(), name)
+ assert Enum.member?(existing_topics(), name)
assert @num_partitions ==
KafkaEx.Protocol.Metadata.Response.partitions_for_topic(

add :uris as a valid option for ConsumerGroup