Apply mix format
Apply mix format to update code quality before enabling github actions checks.
MODIFY
lib/kafka_ex.ex
lib/kafka_ex.ex
@@ -297,11 +297,12 @@ defmodule KafkaEx do
api_version = Keyword.get(opts, :api_version, 0)
# same for offset_commit_api_version
offset_commit_api_version = Keyword.get(opts, :offset_commit_api_version, 0)
+
# By default, it makes sense to synchronize the API version of the offset_commit and the offset_fetch
# operations, otherwise we might commit the offsets in zookeeper and read them from Kafka, meaning
# that the value would be incorrect.
- offset_fetch_api_version = Keyword.get(opts, :offset_fetch_api_version, offset_commit_api_version)
-
+ offset_fetch_api_version =
+ Keyword.get(opts, :offset_fetch_api_version, offset_commit_api_version)
retrieved_offset =
current_offset(

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
@@ -164,9 +164,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
@@ -682,9 +678,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{}}
@@ -735,9 +729,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
@@ -866,9 +858,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}
@@ -895,15 +885,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
@@ -936,9 +922,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)
@@ -960,9 +944,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}

Apply mix format

Apply mix format to update code quality before enabling github actions checks.