Add module to determine API version to use
ADD
lib/kafka_ex/api_versions.ex
+defmodule KafkaEx.ApiVersions do
+
+ def api_versions_map(api_versions) do
+ api_versions
+ |> Enum.map(fn version -> {version.api_key, version} end)
+ |> Map.new
+ end
+
+ def find_api_version(api_versions_map, message_type, {min_implemented_version, max_implemented_version}) do
+ if api_versions_map == [:unsupported] do
+ {:ok, min_implemented_version}
+ else
+ case KafkaEx.Protocol.api_key(message_type) do
+ nil -> :unknown_message_for_client
+ api_key -> case api_versions_map[api_key] do
+ %{min_version: min} when min > max_implemented_version -> :no_version_supported
+ %{max_version: max} when max < min_implemented_version -> :no_version_supported
+ %{max_version: max} -> {:ok, Enum.min([max_implemented_version, max])}
+ _ -> :unknown_message_for_server
+ end
+ end
+ end
+
+ end
+end
\ No newline at end of file

MODIFY
lib/kafka_ex/protocol.ex
lib/kafka_ex/protocol.ex
defmodule KafkaEx.Protocol do
@moduledoc false
- @produce_request 0
- @fetch_request 1
- @offset_request 2
- @metadata_request 3
- @offset_commit_request 8
- @offset_fetch_request 9
- @consumer_metadata_request 10
- @join_group_request 11
- @heartbeat_request 12
- @leave_group_request 13
- @sync_group_request 14
- @api_versions_request 18
- @create_topics_request 19
+ @message_type_to_api_key %{
+ produce: 0,
+ fetch: 1,
+ offset: 2,
+ metadata: 3,
+ offset_commit: 8,
+ offset_fetch: 9,
+ consumer_metadata: 10,
+ join_group: 11,
+ heartbeat: 12,
+ leave_group: 13,
+ sync_group: 14,
+ api_versions: 18,
+ create_topics: 19,
+ }
+
# DescribeConfigs 32
# AlterConfigs 33 Valid resource types are "Topic" and "Broker".
- @api_version 0
-
- defp api_key(:produce) do
- @produce_request
- end
-
- defp api_key(:fetch) do
- @fetch_request
- end
-
- defp api_key(:offset) do
- @offset_request
- end
-
- defp api_key(:metadata) do
- @metadata_request
- end
-
- defp api_key(:offset_commit) do
- @offset_commit_request
- end
-
- defp api_key(:offset_fetch) do
- @offset_fetch_request
- end
-
- defp api_key(:consumer_metadata) do
- @consumer_metadata_request
- end
-
- defp api_key(:join_group) do
- @join_group_request
- end
-
- defp api_key(:heartbeat) do
- @heartbeat_request
- end
-
- defp api_key(:leave_group) do
- @leave_group_request
- end
-
- defp api_key(:sync_group) 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
+ @default_api_version 0
- def create_request(type, correlation_id, client_id) do
- create_request(type, correlation_id, client_id, @api_version)
+ @spec api_key(atom) :: integer | nil
+ def api_key(type) do
+ Map.get(@message_type_to_api_key, type, nil)
end
- def create_request(type, correlation_id, client_id, api_version) do
+ def create_request(type, correlation_id, client_id, api_version \\ @default_api_version) do
<< api_key(type) :: 16, api_version :: 16, correlation_id :: 32,
byte_size(client_id) :: 16, client_id :: binary >>
end

ADD
test/kafka_ex/api_versions_test.exs
+defmodule ApiVersionsTest do
+ use ExUnit.Case
+ @api_versions %{
+ 0 => %KafkaEx.Protocol.ApiVersions.ApiVersion {
+ api_key: 0,
+ min_version: 0,
+ max_version: 3,
+ },
+ 1 => %KafkaEx.Protocol.ApiVersions.ApiVersion {
+ api_key: 1,
+ min_version: 7,
+ max_version: 8,
+ },
+ 3 => %KafkaEx.Protocol.ApiVersions.ApiVersion {
+ api_key: 3,
+ min_version: 2,
+ max_version: 4,
+ }
+ }
+
+ test "can correctly determine the adequate api version when api versions is not support" do
+ assert {:ok, 1} == KafkaEx.ApiVersions.find_api_version([:unsupported], :metadata, {1, 3})
+ end
+
+ test "KafkaEx.ApiVersions can correctly determine the adequate api version when a match exists" do
+ assert {:ok, 3} == KafkaEx.ApiVersions.find_api_version(@api_versions, :metadata, {1, 3})
+ assert {:ok, 0} == KafkaEx.ApiVersions.find_api_version(@api_versions, :produce, {0, 0})
+ end
+
+ test "KafkaEx.ApiVersions replies an error when there's no version match" do
+ assert :no_version_supported == KafkaEx.ApiVersions.find_api_version(@api_versions, :fetch, {0, 6})
+ end
+
+ test "KafkaEx.ApiVersions replies an error when the api_key is unknown to the server" do
+ assert :unknown_message_for_server == KafkaEx.ApiVersions.find_api_version(@api_versions, :create_topics, {0, 1})
+ end
+
+ test "KafkaEx.ApiVersions replies an error when the api_key is unknown to the client" do
+ assert :unknown_message_for_client == KafkaEx.ApiVersions.find_api_version(@api_versions, :this_does_not_exist, {0, 1})
+ end
+end
+

Add module to determine API version to use