Implement ApiVersions message in protocol
MODIFY
lib/kafka_ex.ex
lib/kafka_ex.ex
@@ -529,15 +529,19 @@ defmodule KafkaEx do
def valid_consumer_group?(b) when is_binary(b), do: byte_size(b) > 0
def valid_consumer_group?(_), do: false
-
@doc """
+ Retrieve supported api versions for each api key.
+ """
+ @spec api_versions(Keyword.t) :: CreateTopicsResponse.t
+ def api_versions(opts \\ []) do
+ worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
+ Server.call(worker_name, {:api_versions})
+ end
- ## Example
-
- ```elixir
- iex> KafkaEx.create_worker(:mt)
- ```
+ @doc """
+ Create topics. Must provide a list of CreateTopicsRequest, each containing
+ all the information needed for the creation of a new topic.
"""
@spec create_topics([CreateTopicsRequest.t], Keyword.t) :: CreateTopicsResponse.t
def create_topics(requests, opts \\ []) do

MODIFY
lib/kafka_ex/protocol.ex
lib/kafka_ex/protocol.ex
@@ -12,6 +12,7 @@ defmodule KafkaEx.Protocol do
@heartbeat_request 12
@leave_group_request 13
@sync_group_request 14
+ @api_versions_request 18
@create_topics_request 19
# DescribeConfigs 32
# AlterConfigs 33 Valid resource types are "Topic" and "Broker".
@@ -62,6 +63,10 @@ defmodule KafkaEx.Protocol do
@sync_group_request
end
+ defp api_key(:api_versions) do
+ @api_versions_request
+ end
+
defp api_key(:create_topics) do
@create_topics_request
end

ADD
lib/kafka_ex/protocol/api_versions.ex
+
+defmodule KafkaEx.Protocol.ApiVersions do
+ alias KafkaEx.Protocol
+ require Logger
+
+ # the ApiVersions message can also, itself, have different api versions
+ @default_this_api_version 0
+ @default_throttle_time 0
+
+ @moduledoc """
+ Implementation of the Kafka ApiVersions request and response APIs
+
+ See: https://kafka.apache.org/protocol.html#The_Messages_ApiVersions
+
+ Reference implementation in Java:
+ https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsRequest.java
+ https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
+
+ """
+
+ defmodule Response do
+ @moduledoc false
+ defstruct error_code: nil, api_versions: nil, throttle_time_ms: nil
+ @type t :: %Response{
+ error_code: atom,
+ api_versions: [ApiVersion],
+ throttle_time_ms: integer,
+ }
+ end
+
+ defmodule ApiVersion do
+ @moduledoc false
+ defstruct api_key: nil, min_version: nil, max_version: nil
+ @type t :: %ApiVersion{
+ api_key: integer,
+ min_version: integer,
+ max_version: integer,
+ }
+ end
+
+ @spec create_request(integer, binary, integer) :: binary
+ def create_request(correlation_id, client_id, this_api_version \\ @default_this_api_version)
+
+ def create_request(correlation_id, client_id, 1), do: create_request(correlation_id, client_id, 0)
+ def create_request(correlation_id, client_id, 2), do: create_request(correlation_id, client_id, 0)
+
+ def create_request(correlation_id, client_id, 0) do
+ Protocol.create_request(:api_versions, correlation_id, client_id)
+ end
+
+ @spec parse_response(binary, integer) :: Response.t
+ def parse_response(binary, this_api_version \\ @default_this_api_version)
+
+ def parse_response(<< _correlation_id :: 32-signed,
+ error_code :: 16-signed,
+ api_versions_count :: 32-signed,
+ rest :: binary
+ >>, this_api_version) do
+ Logger.debug("parse_response for api_versions #{this_api_version}")
+
+ %{ parse_rest_of_response(api_versions_count, rest, this_api_version) | error_code: Protocol.error(error_code) }
+ end
+
+ defp parse_rest_of_response(api_versions_count, data, this_api_version) do
+ {api_versions, remaining_data} = Protocol.Common.read_array(api_versions_count, data, &parse_one_api_version/1)
+
+ case {this_api_version, remaining_data} do
+ {0, ""} -> %Response{api_versions: api_versions, throttle_time_ms: @default_throttle_time}
+ {v, << throttle_time_ms :: 32-signed >>} when v in [1, 2] -> %Response{api_versions: api_versions, throttle_time_ms: throttle_time_ms}
+ end
+ end
+
+ defp parse_one_api_version(<< api_key :: 16-signed, min_version :: 16-signed, max_version :: 16-signed, rest :: binary >>) do
+ {%ApiVersion{ api_key: api_key, min_version: min_version, max_version: max_version }, rest}
+ end
+
+end

MODIFY
lib/kafka_ex/protocol/common.ex
lib/kafka_ex/protocol/common.ex
@@ -24,4 +24,15 @@ defmodule KafkaEx.Protocol.Common do
} | parse_topics(topics_size - 1, topics_data, mod)
]
end
+
+ def read_array(0, data_after_array, _read_one) do
+ {[], data_after_array}
+ end
+
+ def read_array(num_items, data, read_one) do
+ {item, rest} = read_one.(data)
+ {items, data_after_array} = read_array(num_items - 1, rest, read_one)
+ {[item | items], data_after_array}
+ end
+
end

MODIFY
lib/kafka_ex/protocol/create_topics.ex
lib/kafka_ex/protocol/create_topics.ex
@@ -4,6 +4,8 @@ defmodule KafkaEx.Protocol.CreateTopics do
@moduledoc """
Implementation of the Kafka CreateTopics request and response APIs
+
+ See: https://kafka.apache.org/protocol.html#The_Messages_CreateTopics
"""
# CreateTopics Request (Version: 0) => [create_topic_requests] timeout

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({:api_versions}, _from, state) do
+ kafka_api_versions(state)
+ end
+
def handle_call({:create_topics, requests, network_timeout}, _from, state) do
kafka_create_topics(requests, network_timeout, state)
end

MODIFY
lib/kafka_ex/server_0_p_10_p_1.ex
lib/kafka_ex/server_0_p_10_p_1.ex
@@ -4,6 +4,7 @@ defmodule KafkaEx.Server0P10P1 do
"""
use KafkaEx.Server
alias KafkaEx.Protocol.CreateTopics
+ alias KafkaEx.Protocol.ApiVersions
alias KafkaEx.Server0P8P2
alias KafkaEx.Server0P9P0
@@ -94,6 +95,15 @@ defmodule KafkaEx.Server0P10P1 do
{:noreply, update_metadata(state, @metadata_api_version)}
end
+ def kafka_api_versions(state) do
+ response = state.correlation_id
+ |> ApiVersions.create_request(@client_id)
+ |> first_broker_response(state)
+ |> ApiVersions.parse_response
+
+ {:reply, response, %{ state | correlation_id: state.correlation_id + 1 }}
+ end
+
def kafka_create_topics(requests, network_timeout, state) do
create_topics_request = %CreateTopics.Request{
create_topic_requests: requests,

MODIFY
lib/kafka_ex/server_0_p_8_p_0.ex
lib/kafka_ex/server_0_p_8_p_0.ex
@@ -63,6 +63,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_api_versions(_state), do: raise "ApiVersions is not supported in 0.8.0 version of kafka"
def kafka_create_topics(_, _, _state), do: raise "CreateTopic is not supported in 0.8.0 version of kafka"
defp fetch(request, state) do

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -134,11 +134,12 @@ defmodule KafkaEx.Server0P8P2 do
{:noreply, state}
end
- def kafka_server_join_group(_, _, _state), do: raise "Join Group is not supported in 0.8.0 version of kafka"
- def kafka_server_sync_group(_, _, _state), do: raise "Sync Group is not supported in 0.8.0 version of kafka"
- 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_create_topics(_, _, _state), do: raise "CreateTopic is not supported in 0.8.0 version of kafka"
+ def kafka_server_join_group(_, _, _state), do: raise "Join Group is not supported in 0.8.2 version of kafka"
+ def kafka_server_sync_group(_, _, _state), do: raise "Sync Group is not supported in 0.8.2 version of kafka"
+ def kafka_server_leave_group(_, _, _state), do: raise "Leave Group is not supported in 0.8.2 version of Kafka"
+ def kafka_server_heartbeat(_, _, _state), do: raise "Heartbeat is not supported in 0.8.2 version of kafka"
+ def kafka_api_versions(_state), do: raise "ApiVersions is not supported in 0.8.2 version of kafka"
+ def kafka_create_topics(_, _, _state), do: raise "CreateTopic is not supported in 0.8.2 version of kafka"
defp update_consumer_metadata(state), do: update_consumer_metadata(state, @retry_count, 0)

MODIFY
lib/kafka_ex/server_0_p_9_p_0.ex
lib/kafka_ex/server_0_p_9_p_0.ex
@@ -45,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
+ defdelegate kafka_api_versions(state), to: Server0P8P2
defdelegate kafka_create_topics(requests, network_timeout, state),to: Server0P8P2
def kafka_server_init([args]) do

MODIFY
test/integration/server0_p_10_p_1_test.exs
test/integration/server0_p_10_p_1_test.exs
@@ -4,7 +4,7 @@ defmodule KafkaEx.Server0P10P1.Test do
@moduletag :server_0_p_10_p_1
- @tag :createtopic
+ @tag :create_topic
test "can create a topic" do
name = "create_topic_#{:rand.uniform(2000000)}"
@@ -40,4 +40,42 @@ defmodule KafkaEx.Server0P10P1.Test do
]} = response
{error_code, topic_name}
end
+
+ @tag :api_version
+ test "can retrieve api versions" do
+
+ # api_key, max_version, min_version
+ api_versions_kafka_0_10_1_0 = [
+ [0, 2, 0],
+ [1, 3, 0],
+ [2, 1, 0],
+ [3, 2, 0],
+ [4, 0, 0],
+ [5, 0, 0],
+ [6, 2, 0],
+ [7, 1, 1],
+ [8, 2, 0],
+ [9, 1, 0],
+ [10, 0, 0],
+ [11, 1, 0],
+ [12, 0, 0],
+ [13, 0, 0],
+ [14, 0, 0],
+ [15, 0, 0],
+ [16, 0, 0],
+ [17, 0, 0],
+ [18, 0, 0],
+ [19, 0, 0],
+ [20, 0, 0]
+ ]
+
+ response = KafkaEx.api_versions()
+ %KafkaEx.Protocol.ApiVersions.Response{
+ api_versions: api_versions,
+ error_code: :no_error,
+ throttle_time_ms: _
+ } = response
+
+ assert api_versions_kafka_0_10_1_0 == api_versions |> Enum.map(&([&1.api_key, &1.max_version, &1.min_version]))
+ end
end

Implement ApiVersions message in protocol