Add preliminary support for CreateTopics API
MODIFY
config/config.exs
config/config.exs
@@ -63,7 +63,7 @@ config :kafka_ex,
],
# set this to the version of the kafka broker that you are using
# include only major.minor.patch versions. must be at least 0.8.0
- kafka_version: "0.9.0"
+ kafka_version: "0.10.1"
env_config = Path.expand("#{Mix.env}.exs", __DIR__)
if File.exists?(env_config) do

MODIFY
docker-compose.yml
docker-compose.yml
@@ -5,32 +5,32 @@ services:
ports:
- "2181:2181"
kafka1:
- image: wurstmeister/kafka:0.9.0.1
+ image: wurstmeister/kafka:0.10.1.0
ports:
- "9092:9092"
depends_on:
- zookeeper
volumes:
- - ./kafka1/server.properties.in:/opt/kafka_2.11-0.9.0.1/config/server.properties.in
+ - ./kafka1/server.properties.in:/opt/kafka_2.11-0.10.1.0/config/server.properties.in
- ./scripts/docker-start-kafka.sh:/usr/bin/start-kafka.sh
- ./ssl:/ssl
kafka2:
- image: wurstmeister/kafka:0.9.0.1
+ image: wurstmeister/kafka:0.10.1.0
ports:
- "9093:9093"
depends_on:
- zookeeper
volumes:
- - ./kafka2/server.properties.in:/opt/kafka_2.11-0.9.0.1/config/server.properties.in
+ - ./kafka2/server.properties.in:/opt/kafka_2.11-0.10.1.0/config/server.properties.in
- ./scripts/docker-start-kafka.sh:/usr/bin/start-kafka.sh
- ./ssl:/ssl
kafka3:
- image: wurstmeister/kafka:0.9.0.1
+ image: wurstmeister/kafka:0.10.1.0
ports:
- "9094:9094"
depends_on:
- zookeeper
volumes:
- - ./kafka3/server.properties.in:/opt/kafka_2.11-0.9.0.1/config/server.properties.in
+ - ./kafka3/server.properties.in:/opt/kafka_2.11-0.10.1.0/config/server.properties.in
- ./scripts/docker-start-kafka.sh:/usr/bin/start-kafka.sh
- ./ssl:/ssl

MODIFY
lib/kafka_ex.ex
lib/kafka_ex.ex
@@ -527,6 +527,23 @@ defmodule KafkaEx do
def valid_consumer_group?(b) when is_binary(b), do: byte_size(b) > 0
def valid_consumer_group?(_), do: false
+
+ @doc """
+
+ ## Example
+
+ ```elixir
+ iex> KafkaEx.create_worker(:mt)
+
+ ```
+ """
+ @spec create_topics(String.t) :: CreateTopics.Response.t
+ def create_topics(topic_name, opts \\ []) do
+ worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
+ # topic = Keyword.get(opts, :topic, "")
+ Server.call(worker_name, {:create_topics, topic_name})
+ end
+
#OTP API
def start(_type, _args) do
max_restarts = Application.get_env(:kafka_ex, :max_restarts, 10)

MODIFY
lib/kafka_ex/config.ex
lib/kafka_ex/config.ex
@@ -80,6 +80,7 @@ defmodule KafkaEx.Config do
defp server("0.8.0"), do: KafkaEx.Server0P8P0
defp server("0.8.2"), do: KafkaEx.Server0P8P2
+ defp server("0.10.1"), do: KafkaEx.Server0P10P1
defp server(_), do: KafkaEx.Server0P9P0
# ssl_options should be an empty list by default if use_ssl is false

MODIFY
lib/kafka_ex/protocol.ex
lib/kafka_ex/protocol.ex
@@ -12,6 +12,9 @@ defmodule KafkaEx.Protocol do
@heartbeat_request 12
@leave_group_request 13
@sync_group_request 14
+ @create_topics_request 19
+ # DescribeConfigs 32
+ # AlterConfigs 33 Valid resource types are "Topic" and "Broker".
@api_version 0
@@ -59,8 +62,16 @@ defmodule KafkaEx.Protocol do
@sync_group_request
end
+ defp api_key(:create_topics) do
+ @create_topics_request
+ end
+
def create_request(type, correlation_id, client_id) do
- << api_key(type) :: 16, @api_version :: 16, correlation_id :: 32,
+ create_request(type, correlation_id, client_id, @api_version)
+ end
+
+ def create_request(type, correlation_id, client_id, api_version) do
+ << api_key(type) :: 16, api_version :: 16, correlation_id :: 32,
byte_size(client_id) :: 16, client_id :: binary >>
end

ADD
lib/kafka_ex/protocol/create_topics.ex
+
+defmodule KafkaEx.Protocol.CreateTopics do
+ alias KafkaEx.Protocol
+
+ @moduledoc """
+ Implementation of the Kafka CreateTopics request and response APIs
+ """
+
+ # CreateTopics Request (Version: 0) => [create_topic_requests] timeout
+ # create_topic_requests => topic num_partitions replication_factor [replica_assignment] [config_entries]
+ # topic => STRING
+ # num_partitions => INT32
+ # replication_factor => INT16
+ # replica_assignment => partition [replicas]
+ # partition => INT32
+ # replicas => INT32
+ # config_entries => config_name config_value
+ # config_name => STRING
+ # config_value => NULLABLE_STRING
+ # timeout => INT32
+
+ defmodule ReplicaAssignment do
+ defstruct partition: nil, replicas: nil
+ @type t :: %ReplicaAssignment{ partition: integer, replicas: [integer] }
+ end
+
+ defmodule ConfigEntry do
+ defstruct config_name: nil, config_value: nil
+ @type t :: %ConfigEntry{ config_name: binary, config_value: binary }
+ end
+
+ defmodule TopicRequest do
+ defstruct topic: nil,
+ num_partitions: nil,
+ replication_factor: nil,
+ replica_assignment: nil,
+ config_entries: nil
+ @type t :: %TopicRequest{
+ topic: binary,
+ num_partitions: integer,
+ replication_factor: integer,
+ replica_assignment: [ReplicaAssignment],
+ config_entries: [ConfigEntry],
+ }
+ end
+
+ defmodule Request do
+ @moduledoc false
+ defstruct create_topic_requests: nil, timeout: nil
+ @type t :: %Request{create_topic_requests: [TopicRequest], timeout: integer}
+ end
+
+ defmodule TopicError do
+ defstruct topic_name: nil, error_code: nil
+ @type t :: %TopicError{ topic_name: binary, error_code: integer }
+ end
+
+ defmodule Response do
+ @moduledoc false
+ defstruct topic_errors: nil
+ @type t :: %Response{topic_errors: [TopicError]}
+ end
+
+ @spec create_request(integer, binary, Request.t) :: binary
+ def create_request(correlation_id, client_id, create_topics_request) do
+ Protocol.create_request(:create_topics, correlation_id, client_id) <>
+ encode_topic_requests(create_topics_request.create_topic_requests) <>
+ << create_topics_request.timeout :: 32-signed >>
+ end
+
+ @spec encode_topic_requests([TopicRequest.t]) :: binary
+ defp encode_topic_requests(requests) do
+ requests
+ |> map_encode(&encode_topic_request/1)
+ end
+
+ @spec encode_topic_request(TopicRequest.t) :: binary
+ defp encode_topic_request(request) do
+ encode_string(request.topic) <>
+ << request.num_partitions :: 32-signed, request.replication_factor :: 16-signed >> <>
+ encode_replica_assignments(request.replica_assignment) <>
+ encode_config_entries(request.config_entries)
+ end
+
+
+ @spec encode_replica_assignments([ReplicaAssignment.t]) :: binary
+ defp encode_replica_assignments(replica_assignments) do
+ replica_assignments |> map_encode(&encode_replica_assignment/1)
+ end
+
+ @spec encode_replica_assignment(ReplicaAssignment.t) :: binary
+ defp encode_replica_assignment(replica_assignment) do
+ << replica_assignment.partition :: 32-signed >> <>
+ replica_assignment.replicas |> map_encode(&(<< &1 :: 32-signed >>))
+ end
+
+ @spec encode_config_entries([ConfigEntry.t]) :: binary
+ defp encode_config_entries(config_entries) do
+ config_entries |> map_encode(&encode_config_entry/1)
+ end
+
+ @spec encode_config_entry(ConfigEntry.t) :: binary
+ defp encode_config_entry(config_entry) do
+ encode_string(config_entry.config_name) <> encode_nullable_string(config_entry.config_value)
+ end
+
+ @spec encode_nullable_string(String.t) :: binary
+ defp encode_nullable_string(text) do
+ if text == nil do
+ << -1 :: 16-signed >>
+ else
+ encode_string(text)
+ end
+ end
+
+ @spec encode_string(String.t) :: binary
+ defp encode_string(text) do
+ << byte_size(text) :: 16-signed, text :: binary, >>
+ end
+
+ defp map_encode(elems, function) do
+ if nil == elems or 0 == length(elems) do
+ << 0 :: 32-signed >>
+ else
+ << length(elems) :: 32-signed >> <>
+ (elems
+ |> Enum.map(function)
+ |> Enum.reduce(&(&1 <> &2)))
+ end
+
+ end
+
+ @spec parse_response(binary) :: [] | Response.t
+ def parse_response(<< _correlation_id :: 32-signed, topic_errors_count :: 32-signed, topic_errors :: binary >>) do
+ %Response{topic_errors: parse_topic_errors(topic_errors_count, topic_errors)}
+ end
+
+ defp parse_topic_errors(0, _), do: []
+
+ @spec parse_topic_errors(integer, binary) :: [TopicError.t]
+ defp parse_topic_errors(topic_errors_count,
+ << topic_name_size :: 16-signed, topic_name :: size(topic_name_size)-binary, error_code :: 16-signed, rest :: binary >>) do
+ [%TopicError{topic_name: topic_name, error_code: error_code} | parse_topic_errors(topic_errors_count - 1, rest)]
+ end
+
+end

MODIFY
lib/kafka_ex/protocol/metadata.ex
lib/kafka_ex/protocol/metadata.ex
@@ -2,6 +2,8 @@ defmodule KafkaEx.Protocol.Metadata do
alias KafkaEx.Protocol
import KafkaEx.Protocol.Common
+ @default_api_version 0
+
@moduledoc """
Implementation of the Kafka Hearbeat request and response APIs
"""
@@ -16,8 +18,8 @@ defmodule KafkaEx.Protocol.Metadata do
alias KafkaEx.Socket
- defstruct node_id: -1, host: "", port: 0, socket: nil
- @type t :: %__MODULE__{}
+ defstruct node_id: -1, host: "", port: 0, socket: nil, is_controller: nil
+ @type t :: %__MODULE__{node_id: integer, host: binary, port: integer, socket: KafkaEx.Socket.t, is_controller: boolean }
def connected?(%Broker{} = broker) do
broker.socket != nil && Socket.open?(broker.socket)
@@ -28,10 +30,11 @@ defmodule KafkaEx.Protocol.Metadata do
@moduledoc false
alias KafkaEx.Protocol.Metadata.Broker
alias KafkaEx.Protocol.Metadata.TopicMetadata
- defstruct brokers: [], topic_metadatas: []
+ defstruct brokers: [], topic_metadatas: [], controller_id: nil
@type t :: %Response{
brokers: [Broker.t],
- topic_metadatas: [TopicMetadata.t]
+ topic_metadatas: [TopicMetadata.t],
+ controller_id: integer
}
def broker_for_topic(metadata, brokers, topic, partition) do
@@ -73,10 +76,11 @@ defmodule KafkaEx.Protocol.Metadata do
defmodule TopicMetadata do
@moduledoc false
alias KafkaEx.Protocol.Metadata.PartitionMetadata
- defstruct error_code: 0, topic: nil, partition_metadatas: []
+ defstruct error_code: 0, topic: nil, is_internal: nil, partition_metadatas: []
@type t :: %TopicMetadata{
error_code: integer | :no_error,
topic: nil | binary,
+ is_internal: nil | boolean,
partition_metadatas: [PartitionMetadata.t]
}
end
@@ -93,19 +97,45 @@ defmodule KafkaEx.Protocol.Metadata do
}
end
- def create_request(correlation_id, client_id, ""), do: KafkaEx.Protocol.create_request(:metadata, correlation_id, client_id) <> << 0 :: 32-signed >>
+ def valid_api_version(v) do
+ case v do
+ nil -> @default_api_version
+ v -> v
+ end
+ end
- def create_request(correlation_id, client_id, topic) when is_binary(topic), do: create_request(correlation_id, client_id, [topic])
+ def create_request(correlation_id, client_id, ""), do: create_request(correlation_id, client_id, "", nil)
+ def create_request(correlation_id, client_id, topics) when is_list(topics), do: create_request(correlation_id, client_id, topics, nil)
+
+ def create_request(correlation_id, client_id, "", api_version) do
+ version = valid_api_version(api_version)
+ topic_count = if 0 == version, do: 0, else: -1
+ KafkaEx.Protocol.create_request(:metadata, correlation_id, client_id, version) <> << topic_count :: 32-signed >>
+ end
+ def create_request(correlation_id, client_id, topic, api_version) when is_binary(topic), do: create_request(correlation_id, client_id, [topic], valid_api_version(api_version))
- def create_request(correlation_id, client_id, topics) when is_list(topics) do
- KafkaEx.Protocol.create_request(:metadata, correlation_id, client_id) <> << length(topics) :: 32-signed, topic_data(topics) :: binary >>
+ def create_request(correlation_id, client_id, topics, api_version) when is_list(topics) do
+ KafkaEx.Protocol.create_request(:metadata, correlation_id, client_id, valid_api_version(api_version)) <> << length(topics) :: 32-signed, topic_data(topics) :: binary >>
end
- def parse_response(<< _correlation_id :: 32-signed, brokers_size :: 32-signed, rest :: binary >>) do
+ def parse_response(data) do
+ parse_response(data, nil)
+ end
+
+ def parse_response(<< _correlation_id :: 32-signed, brokers_size :: 32-signed, rest :: binary >>, api_version) do
+ version = valid_api_version(api_version)
+ case version do
+ 1 ->
+ {brokers, rest} = parse_brokers_v1(brokers_size, rest, [])
+ << controller_id :: 32-signed, rest :: binary >> = rest
+ << topic_metadatas_size :: 32-signed, rest :: binary >> = rest
+ %Response{brokers: brokers, controller_id: controller_id, topic_metadatas: parse_topic_metadatas_v1(topic_metadatas_size, rest)}
+ 0 ->
{brokers, rest} = parse_brokers(brokers_size, rest, [])
<< topic_metadatas_size :: 32-signed, rest :: binary >> = rest
%Response{brokers: brokers, topic_metadatas: parse_topic_metadatas(topic_metadatas_size, rest)}
end
+ end
defp parse_brokers(0, rest, brokers), do: {brokers, rest}
@@ -113,6 +143,32 @@ defmodule KafkaEx.Protocol.Metadata do
parse_brokers(brokers_size - 1, rest, [%Broker{node_id: node_id, host: host, port: port} | brokers])
end
+ defp parse_brokers_v1(0, rest, brokers), do: {brokers, rest}
+
+ defp parse_brokers_v1(brokers_size, <<
+ node_id :: 32-signed,
+ host_len :: 16-signed,
+ host :: size(host_len)-binary,
+ port :: 32-signed,
+ # rack is nullable
+ -1 :: 16-signed,
+ rest :: binary
+ >>, brokers) do
+ parse_brokers_v1(brokers_size - 1, rest, [%Broker{node_id: node_id, host: host, port: port} | brokers])
+ end
+
+ defp parse_brokers_v1(brokers_size, <<
+ node_id :: 32-signed,
+ host_len :: 16-signed,
+ host :: size(host_len)-binary,
+ port :: 32-signed,
+ rack_len :: 16-signed,
+ _rack :: size(rack_len)-binary,
+ rest :: binary
+ >>, brokers) do
+ parse_brokers_v1(brokers_size - 1, rest, [%Broker{node_id: node_id, host: host, port: port} | brokers])
+ end
+
defp parse_topic_metadatas(0, _), do: []
defp parse_topic_metadatas(topic_metadatas_size, << error_code :: 16-signed, topic_len :: 16-signed, topic :: size(topic_len)-binary, partition_metadatas_size :: 32-signed, rest :: binary >>) do
@@ -120,6 +176,22 @@ defmodule KafkaEx.Protocol.Metadata do
[%TopicMetadata{error_code: Protocol.error(error_code), topic: topic, partition_metadatas: partition_metadatas} | parse_topic_metadatas(topic_metadatas_size - 1, rest)]
end
+ defp parse_topic_metadatas_v1(0, _), do: []
+
+ defp parse_topic_metadatas_v1(
+ topic_metadatas_size,
+ << error_code :: 16-signed,
+ topic_len :: 16-signed,
+ topic :: size(topic_len)-binary,
+ # booleans are actually 8-signed
+ is_internal :: 8-signed,
+ partition_metadatas_size :: 32-signed,
+ rest :: binary >>) do
+ {partition_metadatas, rest} = parse_partition_metadatas(partition_metadatas_size, [], rest)
+ [%TopicMetadata{error_code: Protocol.error(error_code), topic: topic, partition_metadatas: partition_metadatas, is_internal: is_internal == 1 } |
+ parse_topic_metadatas_v1(topic_metadatas_size - 1, rest)]
+ end
+
defp parse_partition_metadatas(0, partition_metadatas, rest), do: {partition_metadatas, rest}
defp parse_partition_metadatas(partition_metadatas_size, partition_metadatas, << error_code :: 16-signed, partition_id :: 32-signed, leader :: 32-signed, rest :: binary >>) do

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -260,6 +260,10 @@ defmodule KafkaEx.Server do
kafka_server_heartbeat(request, network_timeout, state)
end
+ def handle_call({:create_topics, topic_name}, _from, state) do
+ kafka_create_topics(topic_name, state)
+ end
+
def handle_info(:update_metadata, state) do
kafka_server_update_metadata(state)
end
@@ -367,9 +371,11 @@ defmodule KafkaEx.Server do
{:noreply, update_metadata(state)}
end
- def update_metadata(state) do
- {correlation_id, metadata} = retrieve_metadata(state.brokers, state.correlation_id, config_sync_timeout())
- metadata_brokers = metadata.brokers
+ def update_metadata(state), do: update_metadata(state, nil)
+
+ def update_metadata(state, api_version) do
+ {correlation_id, metadata} = retrieve_metadata_with_version(state.brokers, state.correlation_id, config_sync_timeout(), api_version)
+ metadata_brokers = metadata.brokers |> Enum.map(&(%{&1 | is_controller: &1.node_id == metadata.controller_id}))
brokers = state.brokers
|> remove_stale_brokers(metadata_brokers)
|> add_new_brokers(metadata_brokers, state.ssl_options, state.use_ssl)
@@ -378,17 +384,27 @@ defmodule KafkaEx.Server do
# credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
def retrieve_metadata(brokers, correlation_id, sync_timeout, topic \\ []), do: retrieve_metadata(brokers, correlation_id, sync_timeout, topic, @retry_count, 0)
+
+ # credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
+ def retrieve_metadata_with_version(brokers, correlation_id, sync_timeout, api_version, topic \\ []), do: retrieve_metadata(brokers, correlation_id, sync_timeout, topic, @retry_count, api_version, 0)
+
# credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
def retrieve_metadata(_, correlation_id, _sync_timeout, topic, 0, error_code) do
Logger.log(:error, "Metadata request for topic #{inspect topic} failed with error_code #{inspect error_code}")
{correlation_id, %Metadata.Response{}}
end
+
+ # credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
+ def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, error_code) do
+ retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, nil, error_code)
+ end
+
# credo:disable-for-next-line Credo.Check.Refactor.FunctionArity
- def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, _error_code) do
- metadata_request = Metadata.create_request(correlation_id, @client_id, topic)
+ def retrieve_metadata(brokers, correlation_id, sync_timeout, topic, retry, api_version, _error_code) do
+ metadata_request = Metadata.create_request(correlation_id, @client_id, topic, api_version)
data = first_broker_response(metadata_request, brokers, sync_timeout)
if data do
- response = Metadata.parse_response(data)
+ response = Metadata.parse_response(data, api_version)
case Enum.find(response.topic_metadatas, &(&1.error_code == :leader_not_available)) do
nil -> {correlation_id + 1, response}

ADD
lib/kafka_ex/server_0_p_10_p_1.ex
+defmodule KafkaEx.Server0P10P1 do
+ @moduledoc """
+ Implements kafkaEx.Server behaviors for kafka 0.10.1 API.
+ """
+ use KafkaEx.Server
+ alias KafkaEx.Protocol.CreateTopics
+ alias KafkaEx.Server0P8P2
+ alias KafkaEx.Server0P9P0
+
+ # alias KafkaEx.ConsumerGroupRequiredError
+ alias KafkaEx.InvalidConsumerGroupError
+ alias KafkaEx.Protocol.ConsumerMetadata
+ alias KafkaEx.Protocol.ConsumerMetadata.Response, as: ConsumerMetadataResponse
+ # alias KafkaEx.Protocol.Heartbeat
+ # alias KafkaEx.Protocol.JoinGroup
+ # alias KafkaEx.Protocol.LeaveGroup
+ alias KafkaEx.Protocol.Metadata.Broker
+ # alias KafkaEx.Protocol.SyncGroup
+ alias KafkaEx.Server.State
+ # alias KafkaEx.NetworkClient
+
+ require Logger
+
+ @metadata_api_version 1
+ @consumer_group_update_interval 30_000
+
+
+ def start_link(args, name \\ __MODULE__)
+
+ def start_link(args, :no_name) do
+ GenServer.start_link(__MODULE__, [args])
+ end
+
+ def start_link(args, name) do
+ GenServer.start_link(__MODULE__, [args, name], [name: name])
+ end
+
+ # The functions below are all defined in KafkaEx.Server0P8P2
+ defdelegate kafka_server_consumer_group(state), to: Server0P8P2
+ defdelegate kafka_server_fetch(fetch_request, state), to: Server0P8P2
+ defdelegate kafka_server_offset_fetch(offset_fetch, state), to: Server0P8P2
+ defdelegate kafka_server_offset_commit(offset_commit_request, state), to: Server0P8P2
+ defdelegate kafka_server_consumer_group_metadata(state), to: Server0P8P2
+ defdelegate kafka_server_update_consumer_metadata(state), to: Server0P8P2
+
+ # The functions below are all defined in KafkaEx.Server0P9P0
+ defdelegate kafka_server_join_group(request, network_timeout, state_in), to: Server0P9P0
+ defdelegate kafka_server_sync_group(request, network_timeout, state_in), to: Server0P9P0
+ defdelegate kafka_server_leave_group(request, network_timeout, state_in), to: Server0P9P0
+ defdelegate kafka_server_heartbeat(request, network_timeout, state_in), to: Server0P9P0
+ defdelegate consumer_group?(state), to: Server0P9P0
+
+ def kafka_server_init([args]) do
+ kafka_server_init([args, self()])
+ end
+
+ def kafka_server_init([args, name]) do
+ uris = Keyword.get(args, :uris, [])
+ metadata_update_interval = Keyword.get(args, :metadata_update_interval, @metadata_update_interval)
+ consumer_group_update_interval = Keyword.get(args, :consumer_group_update_interval, @consumer_group_update_interval)
+
+ # this should have already been validated, but it's possible someone could
+ # try to short-circuit the start call
+ consumer_group = Keyword.get(args, :consumer_group)
+ unless KafkaEx.valid_consumer_group?(consumer_group) do
+ raise InvalidConsumerGroupError, consumer_group
+ end
+
+ use_ssl = Keyword.get(args, :use_ssl, false)
+ ssl_options = Keyword.get(args, :ssl_options, [])
+
+ brokers = Enum.map(uris, fn({host, port}) -> %Broker{host: host, port: port, socket: NetworkClient.create_socket(host, port, ssl_options, use_ssl)} end)
+
+ {correlation_id, metadata} = retrieve_metadata_with_version(brokers, 0, config_sync_timeout(), @metadata_api_version)
+ state = %State{metadata: metadata, brokers: brokers, correlation_id: correlation_id, consumer_group: consumer_group, metadata_update_interval: metadata_update_interval, consumer_group_update_interval: consumer_group_update_interval, worker_name: name, ssl_options: ssl_options, use_ssl: use_ssl}
+ # Get the initial "real" broker list and start a regular refresh cycle.
+ state = update_metadata(state, @metadata_api_version)
+ {:ok, _} = :timer.send_interval(state.metadata_update_interval, :update_metadata)
+
+ state =
+ if consumer_group?(state) do
+ # If we are using consumer groups then initialize the state and start the update cycle
+ {_, updated_state} = update_consumer_metadata(state)
+ {:ok, _} = :timer.send_interval(state.consumer_group_update_interval, :update_consumer_metadata)
+ updated_state
+ else
+ state
+ end
+
+ {:ok, state}
+ end
+
+ def kafka_server_metadata(topic, state) do
+ {correlation_id, metadata} = retrieve_metadata_with_version(state.brokers, state.correlation_id, config_sync_timeout(), @metadata_api_version, topic)
+ updated_state = %{state | metadata: metadata, correlation_id: correlation_id}
+ {:reply, metadata, updated_state}
+ end
+
+ def kafka_server_update_metadata(state) do
+ {:noreply, update_metadata(state, @metadata_api_version)}
+ end
+
+ def kafka_create_topics(requests, state) do
+ create_topics_request = %CreateTopics.Request{
+ create_topic_requests: requests,
+ timeout: 2000
+ }
+
+ mainRequest = CreateTopics.create_request(state.correlation_id, @client_id, create_topics_request)
+
+ broker = state.brokers |> Enum.find(&(&1.is_controller))
+
+ {response, state} = case broker do
+ nil ->
+ Logger.log(:error, "Coordinator for topic is not available")
+ {:topic_not_found, state}
+ _ ->
+ response = broker
+ |> NetworkClient.send_sync_request(mainRequest, config_sync_timeout())
+ |> case do
+ {:error, reason} -> {:error, reason}
+ response -> CreateTopics.parse_response(response)
+ end
+ {response, %{state | correlation_id: state.correlation_id + 1}}
+ end
+
+ {:reply, response, state}
+ end
+
+ defp update_consumer_metadata(state), do: update_consumer_metadata(state, @retry_count, 0)
+
+ defp update_consumer_metadata(%State{consumer_group: consumer_group} = state, 0, error_code) do
+ Logger.log(:error, "Fetching consumer_group #{consumer_group} metadata failed with error_code #{inspect error_code}")
+ {%ConsumerMetadataResponse{error_code: error_code}, state}
+ end
+
+ defp update_consumer_metadata(%State{consumer_group: consumer_group, correlation_id: correlation_id} = state, retry, _error_code) do
+ response = correlation_id
+ |> ConsumerMetadata.create_request(@client_id, consumer_group)
+ |> first_broker_response(state)
+ |> ConsumerMetadata.parse_response
+
+ case response.error_code do
+ :no_error ->
+ {
+ response,
+ %{
+ state |
+ consumer_metadata: response,
+ correlation_id: state.correlation_id + 1
+ }
+ }
+ _ -> :timer.sleep(400)
+ update_consumer_metadata(
+ %{state | correlation_id: state.correlation_id + 1},
+ retry - 1,
+ response.error_code
+ )
+ end
+ end
+
+ defp broker_for_consumer_group(state) do
+ ConsumerMetadataResponse.broker_for_consumer_group(state.brokers, state.consumer_metadata)
+ end
+
+ # refactored from two versions, one that used the first broker as valid answer, hence
+ # the optional extra flag to do that. Wraps broker_for_consumer_group with an update
+ # call if no broker was found.
+ def broker_for_consumer_group_with_update(state, use_first_as_default \\ false) do
+ case broker_for_consumer_group(state) do
+ nil ->
+ {_, updated_state} = update_consumer_metadata(state)
+ default_broker = if use_first_as_default, do: hd(state.brokers), else: nil
+ {broker_for_consumer_group(updated_state) || default_broker, updated_state}
+ broker ->
+ {broker, state}
+ end
+ end
+
+ defp first_broker_response(request, state) do
+ first_broker_response(request, state.brokers, config_sync_timeout())
+ end
+
+end

MODIFY
lib/kafka_ex/server_0_p_8_p_0.ex
lib/kafka_ex/server_0_p_8_p_0.ex
@@ -62,6 +62,7 @@ defmodule KafkaEx.Server0P8P0 do
def kafka_server_leave_group(_, _, _state), do: raise "Leave Group is not supported in 0.8.0 version of Kafka"
def kafka_server_heartbeat(_, _, _state), do: raise "Heartbeat is not supported in 0.8.0 version of kafka"
def kafka_server_update_consumer_metadata(_state), do: raise "Consumer Group Metadata is not supported in 0.8.0 version of kafka"
+ def kafka_create_topics(_, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
defp fetch(request, state) do
case network_request(request, Fetch, state) do

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -138,6 +138,7 @@ defmodule KafkaEx.Server0P8P2 do
def kafka_server_leave_group(_, _, _state), do: raise "Leave Group is not supported in 0.8.0 version of Kafka"
def kafka_server_heartbeat(_, _, _state), do: raise "Heartbeat is not supported in 0.8.0 version of kafka"
defp update_consumer_metadata(state), do: update_consumer_metadata(state, @retry_count, 0)
+ def kafka_create_topics(_, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
defp update_consumer_metadata(%State{consumer_group: consumer_group} = state, 0, error_code) do
Logger.log(:error, "Fetching consumer_group #{consumer_group} metadata failed with error_code #{inspect error_code}")

MODIFY
lib/kafka_ex/server_0_p_9_p_0.ex
lib/kafka_ex/server_0_p_9_p_0.ex
defmodule KafkaEx.Server0P9P0 do
@moduledoc """
- Implements kafkaEx.Server behaviors for kafka 0.9.0 API.
+ Implements kafkaEx.Server behaviors for kafka >= 0.9.0 < 0.10.1 API.
+
"""
+
+ # these functions aren't implemented for 0.9.0
+ @dialyzer [
+ {:nowarn_function, kafka_create_topics: 2},
+ ]
+
use KafkaEx.Server
alias KafkaEx.ConsumerGroupRequiredError
alias KafkaEx.InvalidConsumerGroupError
@@ -38,6 +45,7 @@ defmodule KafkaEx.Server0P9P0 do
defdelegate kafka_server_offset_commit(offset_commit_request, state), to: Server0P8P2
defdelegate kafka_server_consumer_group_metadata(state), to: Server0P8P2
defdelegate kafka_server_update_consumer_metadata(state), to: Server0P8P2
+ def kafka_create_topics(_, _state), do: raise "CreateTopic is not supported in 0.9.0 version of kafka"
def kafka_server_init([args]) do
kafka_server_init([args, self()])
@@ -207,7 +215,7 @@ defmodule KafkaEx.Server0P9P0 do
# refactored from two versions, one that used the first broker as valid answer, hence
# the optional extra flag to do that. Wraps broker_for_consumer_group with an update
# call if no broker was found.
- defp broker_for_consumer_group_with_update(state, use_first_as_default \\ false) do
+ def broker_for_consumer_group_with_update(state, use_first_as_default \\ false) do
case broker_for_consumer_group(state) do
nil ->
{_, updated_state} = update_consumer_metadata(state)
@@ -224,6 +232,7 @@ defmodule KafkaEx.Server0P9P0 do
def consumer_group?(%State{consumer_group: :no_consumer_group}), do: false
def consumer_group?(_), do: true
+
defp first_broker_response(request, state) do
first_broker_response(request, state.brokers, config_sync_timeout())
end

MODIFY
scripts/docker_up.sh
scripts/docker_up.sh
@@ -14,6 +14,7 @@ set -e
# Kafka needs to know our ip address so that it can advertise valid
# connnection details
+
if [ -z ${IP_IFACE} ]
then
echo Detecting active network interface

ADD
test/integration/server0_p_10_p_1_test.exs
+defmodule KafkaEx.Server0P10P1.Test do
+ use ExUnit.Case
+
+ @moduletag :server_0_p_10_p_1
+
+ @tag :createtopic
+ test "can create a topic" do
+ name = "topic3_#{:rand.uniform(2000000)}"
+
+ request = %{
+ topic: name,
+ num_partitions: 10,
+ replication_factor: 1,
+ replica_assignment: [],
+ config_entries: [
+ %{config_name: "cleanup.policy", config_value: "compact"},
+ %{config_name: "min.compaction.lag.ms", config_value: "0"}
+ ]}
+
+ resp = KafkaEx.create_topics([request])
+ # error = NONE
+ assert {0, name} == parse_create_topic_resp(resp)
+
+ resp = KafkaEx.create_topics([request])
+ # error = TOPIC_ALREADY_EXISTS
+ assert {36, name} == parse_create_topic_resp(resp)
+
+ topics = KafkaEx.metadata.topic_metadatas |> Enum.map(&(&1.topic))
+ assert Enum.member?(topics, name)
+ end
+
+ def parse_create_topic_resp(response) do
+ %KafkaEx.Protocol.CreateTopics.Response{
+ topic_errors: [
+ %KafkaEx.Protocol.CreateTopics.TopicError{
+ error_code: error_code,
+ topic_name: topic_name
+ }
+ ]} = response
+ {error_code, topic_name}
+ end
+end

Add preliminary support for CreateTopics API