MInor fixes
Update ssl related code and nest it into describes
Updates tests to support new broker version
Update docker-compose to create topics
fix check
MODIFY
.github/workflows/checks.yml
.github/workflows/checks.yml
@@ -31,39 +31,43 @@ jobs:
elixir-version: ${{matrix.elixir}} # Define the elixir version [required]
otp-version: ${{matrix.erlang}} # Define the OTP version [required]
experimental-otp: true # More info https://github.com/actions/setup-elixir/issues/31
- - name: Retrieve cached dependencies
+
+ - name: Retrieve Mix Dependencies Cache
uses: actions/cache@v2
- id: mix-cache
+ id: mix-cache # id to use in retrieve action
with:
- path: |
- deps
- _build
- key: ${{ runner.os }}-${{ matrix.otp }}-${{ matrix.elixir }}-${{ hashFiles('mix.lock') }}
- - name: Install dependencies
+ path: deps
+ key: ${{ runner.os }}-${{ matrix.otp }}-${{ matrix.elixir }}-mix-${{ hashFiles(format('{0}{1}', github.workspace, '/mix.lock')) }}
+
+ - name: Retrieve Mix Dependencies Compilation Cache
+ uses: actions/cache@v2
+ id: mix-deps-compile-cache # id to use in retrieve action
+ with:
+ path: _build
+ key: ${{ runner.os }}-${{ matrix.otp }}-${{ matrix.elixir }}-mix-deps-compile-${{ hashFiles(format('{0}{1}', github.workspace, '/mix.lock')) }}
+ - name: Install Mix Dependencies
if: steps.mix-cache.outputs.cache-hit != 'true'
run: |
mix local.rebar --force
mix local.hex --force
mix deps.get
- mix deps.compile
+
+ - name: Compile Mix Dependencies
+ if: steps.mix-deps-compile-cache.outputs.cache-hit != 'true'
+ run: mix deps.compile
run-checks:
name: check
runs-on: ubuntu-20.04
- needs: [setup]
+ needs: [dependencies]
env:
MIX_ENV: dev
strategy:
fail-fast: false
matrix:
- include:
- - elixir: 1.10.2
- erlang: 22.3
- checks:
- - mix credo
- - mix compile --warnings-as-errors --force
- - mix format --check-formatted
+ elixir: ['1.12.3']
+ erlang: ['24.3']
steps:
- uses: actions/checkout@v2
@@ -88,20 +92,22 @@ jobs:
path: _build
key: ${{ runner.os }}-${{ matrix.otp }}-${{ matrix.elixir }}-mix-deps-compile-${{ hashFiles(format('{0}{1}', github.workspace, '/mix.lock')) }}
- - run: ${{ matrix.checks }}
+ - run: |
+ mix compile --force
+ mix credo
+ mix format --check-formatted
dialyzer:
name: check / dialyzer
- needs: [setup]
+ needs: [dependencies]
runs-on: ubuntu-latest
env:
MIX_ENV: dev
strategy:
matrix:
- include:
- - elixir: 1.10.2
- erlang: 22.3
+ elixir: ['1.12.3']
+ erlang: ['24.3']
steps:
- uses: actions/checkout@v2

MODIFY
docker-compose.yml
docker-compose.yml
@@ -69,9 +69,8 @@ services:
- ./ssl/kafka.server.keystore.jks:/etc/kafka/secrets/kafka.server.keystore.jks:ro,z
- ./ssl/kafka.server.truststore.jks:/etc/kafka/secrets/kafka.server.truststore.jks:ro,z
- ready:
- image: busybox:1.31-musl
- command: tail -f /dev/null
+ kafka_setup:
+ image: confluentinc/cp-kafka:5.5.1
depends_on:
kafka-1:
condition: service_healthy
@@ -79,3 +78,6 @@ services:
condition: service_healthy
kafka-3:
condition: service_healthy
+ command: "bash -c 'kafka-topics --zookeeper zookeeper:32181 --create --if-not-exists --partitions 4 --replication-factor 2 --topic consumer_group_implementation_test && \
+ kafka-topics --zookeeper zookeeper:32181 --create --if-not-exists --partitions 4 --replication-factor 2 --topic test0p8p0 && \
+ kafka-topics --zookeeper zookeeper:32181 --list'"

MODIFY
lib/kafka_ex/gen_consumer.ex
lib/kafka_ex/gen_consumer.ex
@@ -810,9 +810,7 @@ defmodule KafkaEx.GenConsumer do
KafkaEx.latest_offset(topic, partition, worker_name)
_ ->
- raise "Offset out of range while consuming topic #{topic}, partition #{
- partition
- }."
+ raise "Offset out of range while consuming topic #{topic}, partition #{partition}."
end
%State{

MODIFY
lib/kafka_ex/network_client.ex
lib/kafka_ex/network_client.ex
@@ -25,9 +25,7 @@ defmodule KafkaEx.NetworkClient do
err ->
Logger.log(
:error,
- "Could not connect to broker #{inspect(host)}:#{inspect(port)} because of error #{
- inspect(err)
- }"
+ "Could not connect to broker #{inspect(host)}:#{inspect(port)} because of error #{inspect(err)}"
)
nil
@@ -50,9 +48,7 @@ defmodule KafkaEx.NetworkClient do
{_, reason} ->
Logger.log(
:error,
- "Asynchronously sending data to broker #{inspect(broker.host)}:#{
- inspect(broker.port)
- } failed with #{inspect(reason)}"
+ "Asynchronously sending data to broker #{inspect(broker.host)}:#{inspect(broker.port)} failed with #{inspect(reason)}"
)
reason
@@ -77,9 +73,7 @@ defmodule KafkaEx.NetworkClient do
{:error, reason} ->
Logger.log(
:error,
- "Receiving data from broker #{inspect(broker.host)}:#{
- inspect(broker.port)
- } failed with #{inspect(reason)}"
+ "Receiving data from broker #{inspect(broker.host)}:#{inspect(broker.port)} failed with #{inspect(reason)}"
)
Socket.close(socket)
@@ -90,9 +84,7 @@ defmodule KafkaEx.NetworkClient do
{_, reason} ->
Logger.log(
:error,
- "Sending data to broker #{inspect(broker.host)}:#{
- inspect(broker.port)
- } failed with #{inspect(reason)}"
+ "Sending data to broker #{inspect(broker.host)}:#{inspect(broker.port)} failed with #{inspect(reason)}"
)
Socket.close(socket)

MODIFY
lib/kafka_ex/protocol/fetch.ex
lib/kafka_ex/protocol/fetch.ex
@@ -162,9 +162,7 @@ defmodule KafkaEx.Protocol.Fetch do
)
when byte_size(partial_message_data) < msg_size do
raise RuntimeError,
- "Insufficient data fetched at offset #{offset}. Message size is #{
- msg_size
- } but only received #{byte_size(partial_message_data)} bytes. Try increasing max_bytes."
+ "Insufficient data fetched at offset #{offset}. Message size is #{msg_size} but only received #{byte_size(partial_message_data)} bytes. Try increasing max_bytes."
end
# handles the single message case and the batch (compression) case

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -373,9 +373,7 @@ defmodule KafkaEx.Server do
def terminate(reason, state) do
Logger.log(
:debug,
- "Shutting down worker #{inspect(state.worker_name)}, reason: #{
- inspect(reason)
- }"
+ "Shutting down worker #{inspect(state.worker_name)}, reason: #{inspect(reason)}"
)
if state.event_pid do
@@ -471,9 +469,7 @@ defmodule KafkaEx.Server do
nil ->
Logger.log(
:error,
- "kafka_server_produce_send_request: leader for topic #{
- produce_request.topic
- }/#{produce_request.partition} is not available"
+ "kafka_server_produce_send_request: leader for topic #{produce_request.topic}/#{produce_request.partition} is not available"
)
:leader_not_available
@@ -678,9 +674,7 @@ defmodule KafkaEx.Server do
) do
Logger.log(
:error,
- "Metadata request for topic #{inspect(topic)} failed with error_code #{
- inspect(error_code)
- }"
+ "Metadata request for topic #{inspect(topic)} failed with error_code #{inspect(error_code)}"
)
{correlation_id, %Metadata.Response{}}
@@ -731,9 +725,7 @@ defmodule KafkaEx.Server do
end
else
message =
- "Unable to fetch metadata from any brokers. Timeout is #{
- sync_timeout
- }."
+ "Unable to fetch metadata from any brokers. Timeout is #{sync_timeout}."
Logger.log(:error, message)
raise message
@@ -862,9 +854,7 @@ defmodule KafkaEx.Server do
case broker do
nil ->
Logger.error(fn ->
- "network_request: leader for topic #{request.topic}/#{
- request.partition
- } is not available"
+ "network_request: leader for topic #{request.topic}/#{request.partition} is not available"
end)
{{:error, :topic_not_found}, updated_state}
@@ -891,15 +881,11 @@ defmodule KafkaEx.Server do
rescue
_ ->
Logger.error(
- "Failed to parse a response from the server: #{
- inspect(response)
- }"
+ "Failed to parse a response from the server: #{inspect(response)}"
)
Kernel.reraise(
- "Parse error during #{inspect(module)}.parse_response. Couldn't parse: #{
- inspect(response)
- }",
+ "Parse error during #{inspect(module)}.parse_response. Couldn't parse: #{inspect(response)}",
System.stacktrace()
)
end
@@ -932,9 +918,7 @@ defmodule KafkaEx.Server do
Enum.each(brokers_to_remove, fn broker ->
Logger.log(
:debug,
- "Closing connection to broker #{broker.node_id}: #{
- inspect(broker.host)
- } on port #{inspect(broker.port)}"
+ "Closing connection to broker #{broker.node_id}: #{inspect(broker.host)} on port #{inspect(broker.port)}"
)
NetworkClient.close_socket(broker.socket)
@@ -956,9 +940,7 @@ defmodule KafkaEx.Server do
nil ->
Logger.log(
:debug,
- "Establishing connection to broker #{metadata_broker.node_id}: #{
- inspect(metadata_broker.host)
- } on port #{inspect(metadata_broker.port)}"
+ "Establishing connection to broker #{metadata_broker.node_id}: #{inspect(metadata_broker.host)} on port #{inspect(metadata_broker.port)}"
)
add_new_brokers(

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -226,9 +226,7 @@ defmodule KafkaEx.Server0P8P2 do
) do
Logger.log(
:error,
- "Fetching consumer_group #{consumer_group} metadata failed with error_code #{
- inspect(error_code)
- }"
+ "Fetching consumer_group #{consumer_group} metadata failed with error_code #{inspect(error_code)}"
)
{%ConsumerMetadataResponse{error_code: error_code}, state}

MODIFY
mix.exs
mix.exs
@@ -42,7 +42,7 @@ defmodule KafkaEx.Mixfile do
main_deps = [
{:kayrock, "~> 0.1.12"},
{:credo, "~> 1.1", only: [:dev, :test], runtime: false},
- {:dialyxir, "~> 1.0.0-rc.3", only: :dev, runtime: false},
+ {:dialyxir, "~> 1.0", only: :dev, runtime: false},
{:excoveralls, "~> 0.7", only: :test, runtime: false},
{:snappy,
git: "https://github.com/fdmanana/snappy-erlang-nif", only: [:dev, :test]}

MODIFY
mix.lock
mix.lock
"connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm", "4a0850c9be22a43af9920a71ab17c051f5f7d45c209e40269a1938832510e4d9"},
"crc32cer": {:hex, :crc32cer, "0.1.10", "fb87abbf34b72f180f8c3a908cd1826c6cb9a59787d156a29e05de9e98be385e", [:rebar3], [], "hexpm", "5b1f47efd0a1b4b7411f1f35e14d3c8c6da6e6a2a725ec8f2cf1ab13703e5f38"},
"credo": {:hex, :credo, "1.1.5", "caec7a3cadd2e58609d7ee25b3931b129e739e070539ad1a0cd7efeeb47014f4", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "d0bbd3222607ccaaac5c0340f7f525c627ae4d7aee6c8c8c108922620c5b6446"},
- "dialyxir": {:hex, :dialyxir, "1.0.0-rc.7", "6287f8f2cb45df8584317a4be1075b8c9b8a69de8eeb82b4d9e6c761cf2664cd", [:mix], [{:erlex, ">= 0.2.5", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "506294d6c543e4e5282d4852aead19ace8a35bedeb043f9256a06a6336827122"},
+ "dialyxir": {:hex, :dialyxir, "1.1.0", "c5aab0d6e71e5522e77beff7ba9e08f8e02bad90dfbeffae60eaf0cb47e29488", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "07ea8e49c45f15264ebe6d5b93799d4dd56a44036cf42d0ad9c960bc266c0b9a"},
"earmark": {:hex, :earmark, "1.4.3", "364ca2e9710f6bff494117dbbd53880d84bebb692dafc3a78eb50aa3183f2bfd", [:mix], [], "hexpm", "8cf8a291ebf1c7b9539e3cddb19e9cef066c2441b1640f13c34c1d3cfc825fec"},
- "erlex": {:hex, :erlex, "0.2.5", "e51132f2f472e13d606d808f0574508eeea2030d487fc002b46ad97e738b0510", [:mix], [], "hexpm", "756d3e19b056339af674b715fdd752c5dac468cf9d0e2d1a03abf4574e99fbf8"},
+ "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
"ex_doc": {:hex, :ex_doc, "0.21.2", "caca5bc28ed7b3bdc0b662f8afe2bee1eedb5c3cf7b322feeeb7c6ebbde089d6", [:mix], [{:earmark, "~> 1.3.3 or ~> 1.4", [hex: :earmark, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}], "hexpm", "f1155337ae17ff7a1255217b4c1ceefcd1860b7ceb1a1874031e7a861b052e39"},
"excoveralls": {:hex, :excoveralls, "0.12.1", "a553c59f6850d0aff3770e4729515762ba7c8e41eedde03208182a8dc9d0ce07", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "5c1f717066a299b1b732249e736c5da96bb4120d1e55dc2e6f442d251e18a812"},
"hackney": {:hex, :hackney, "1.15.2", "07e33c794f8f8964ee86cebec1a8ed88db5070e52e904b8f12209773c1036085", [:rebar3], [{:certifi, "2.5.1", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "6.0.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.5", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm", "e0100f8ef7d1124222c11ad362c857d3df7cb5f4204054f9f0f4a728666591fc"},

MODIFY
test/integration/kayrock/compatibility_0_p_10_and_later_test.exs
test/integration/kayrock/compatibility_0_p_10_and_later_test.exs
@@ -65,40 +65,54 @@ defmodule KafkaEx.KayrockCompatibility0p10AndLaterTest do
# note this checks against the version of broker we're running in test
# api_key, max_version, min_version
api_versions_kafka_0_11_0_1 = [
- [0, 3, 0],
- [1, 5, 0],
- [2, 2, 0],
- [3, 4, 0],
- [4, 0, 0],
- [5, 0, 0],
- [6, 3, 0],
- [7, 1, 1],
- [8, 3, 0],
- [9, 3, 0],
- [10, 1, 0],
- [11, 2, 0],
- [12, 1, 0],
- [13, 1, 0],
- [14, 1, 0],
- [15, 1, 0],
- [16, 1, 0],
- [17, 0, 0],
- [18, 1, 0],
- [19, 2, 0],
- [20, 1, 0],
- [21, 0, 0],
- [22, 0, 0],
- [23, 0, 0],
- [24, 0, 0],
- [25, 0, 0],
- [26, 0, 0],
+ [0, 8, 0],
+ [1, 11, 0],
+ [2, 5, 0],
+ [3, 9, 0],
+ [4, 4, 0],
+ [5, 2, 0],
+ [6, 6, 0],
+ [7, 3, 0],
+ [8, 8, 0],
+ [9, 7, 0],
+ [10, 3, 0],
+ [11, 7, 0],
+ [12, 4, 0],
+ [13, 4, 0],
+ [14, 5, 0],
+ [15, 5, 0],
+ [16, 3, 0],
+ [17, 1, 0],
+ [18, 3, 0],
+ [19, 5, 0],
+ [20, 4, 0],
+ [21, 1, 0],
+ [22, 3, 0],
+ [23, 3, 0],
+ [24, 1, 0],
+ [25, 1, 0],
+ [26, 1, 0],
[27, 0, 0],
- [28, 0, 0],
- [29, 0, 0],
- [30, 0, 0],
- [31, 0, 0],
- [32, 0, 0],
- [33, 0, 0]
+ [28, 3, 0],
+ [29, 2, 0],
+ [30, 2, 0],
+ [31, 2, 0],
+ [32, 2, 0],
+ [33, 1, 0],
+ [34, 1, 0],
+ [35, 1, 0],
+ [36, 2, 0],
+ [37, 2, 0],
+ [38, 2, 0],
+ [39, 2, 0],
+ [40, 2, 0],
+ [41, 2, 0],
+ [42, 2, 0],
+ [43, 2, 0],
+ [44, 1, 0],
+ [45, 0, 0],
+ [46, 0, 0],
+ [47, 0, 0]
]
response = KafkaEx.api_versions(worker_name: client)

MODIFY
test/integration/server0_p_10_p_1_test.exs
test/integration/server0_p_10_p_1_test.exs
@@ -11,40 +11,54 @@ defmodule KafkaEx.Server0P10P1.Test do
# note this checks against the version of broker we're running in test
# api_key, max_version, min_version
api_versions_kafka_0_11_0_1 = [
- [0, 3, 0],
- [1, 5, 0],
- [2, 2, 0],
- [3, 4, 0],
- [4, 0, 0],
- [5, 0, 0],
- [6, 3, 0],
- [7, 1, 1],
- [8, 3, 0],
- [9, 3, 0],
- [10, 1, 0],
- [11, 2, 0],
- [12, 1, 0],
- [13, 1, 0],
- [14, 1, 0],
- [15, 1, 0],
- [16, 1, 0],
- [17, 0, 0],
- [18, 1, 0],
- [19, 2, 0],
- [20, 1, 0],
- [21, 0, 0],
- [22, 0, 0],
- [23, 0, 0],
- [24, 0, 0],
- [25, 0, 0],
- [26, 0, 0],
+ [0, 8, 0],
+ [1, 11, 0],
+ [2, 5, 0],
+ [3, 9, 0],
+ [4, 4, 0],
+ [5, 2, 0],
+ [6, 6, 0],
+ [7, 3, 0],
+ [8, 8, 0],
+ [9, 7, 0],
+ [10, 3, 0],
+ [11, 7, 0],
+ [12, 4, 0],
+ [13, 4, 0],
+ [14, 5, 0],
+ [15, 5, 0],
+ [16, 3, 0],
+ [17, 1, 0],
+ [18, 3, 0],
+ [19, 5, 0],
+ [20, 4, 0],
+ [21, 1, 0],
+ [22, 3, 0],
+ [23, 3, 0],
+ [24, 1, 0],
+ [25, 1, 0],
+ [26, 1, 0],
[27, 0, 0],
- [28, 0, 0],
- [29, 0, 0],
- [30, 0, 0],
- [31, 0, 0],
- [32, 0, 0],
- [33, 0, 0]
+ [28, 3, 0],
+ [29, 2, 0],
+ [30, 2, 0],
+ [31, 2, 0],
+ [32, 2, 0],
+ [33, 1, 0],
+ [34, 1, 0],
+ [35, 1, 0],
+ [36, 2, 0],
+ [37, 2, 0],
+ [38, 2, 0],
+ [39, 2, 0],
+ [40, 2, 0],
+ [41, 2, 0],
+ [42, 2, 0],
+ [43, 2, 0],
+ [44, 1, 0],
+ [45, 0, 0],
+ [46, 0, 0],
+ [47, 0, 0]
]
response = KafkaEx.api_versions()

MODIFY
test/socket_test.exs
test/socket_test.exs
@@ -36,6 +36,7 @@ defmodule KafkaEx.Socket.Test do
{:ok, listen_socket} =
:ssl.listen(port, [
:binary,
+ {:verify, :verify_none},
{:active, false},
{:reuseaddr, true},
{:packet, 0},
@@ -49,7 +50,7 @@ defmodule KafkaEx.Socket.Test do
defp listen(socket) do
case :ssl.transport_accept(socket) do
{:ok, conn} ->
- :ok = :ssl.ssl_accept(conn)
+ {:ok, _socket} = :ssl.handshake(conn)
pid = spawn_link(fn -> recv(conn) end)
:ssl.controlling_process(socket, pid)
@@ -73,9 +74,12 @@ defmodule KafkaEx.Socket.Test do
setup_all do
:ssl.start()
- SSLServer.start(3030)
+ end
+
+ describe "without SSL socket" do
+ setup do
Server.start(3040)
- {:ok, [ssl_port: 3030, port: 3040]}
+ {:ok, [port: 3040]}
end
test "create a non SSL socket", context do
@@ -119,6 +123,13 @@ defmodule KafkaEx.Socket.Test do
KafkaEx.Socket.close(socket)
assert {:error, :closed} == KafkaEx.Socket.send(socket, 'ping')
end
+ end
+
+ describe "with ssl socket" do
+ setup do
+ SSLServer.start(3030)
+ {:ok, [ssl_port: 3030]}
+ end
test "create a SSL socket", context do
{:ok, socket} =
@@ -162,3 +173,4 @@ defmodule KafkaEx.Socket.Test do
assert {:error, :closed} == KafkaEx.Socket.send(socket, 'ping')
end
end
+end

MInor fixes

- Update ssl related code and nest it into describes
- Updates tests to support new broker version
- Update docker-compose to create topics
- fix check