Prevent memory leak in message producer when request is invalid
MODIFY
lib/kafka_ex/protocol/produce.ex
lib/kafka_ex/protocol/produce.ex
@@ -82,12 +82,16 @@ defmodule KafkaEx.Protocol.Produce do
def parse_response(unknown), do: unknown
defp create_message_set([], _compression_type), do: {"", 0}
- defp create_message_set([%Message{key: key, value: value}|messages], :none) do
+ defp create_message_set_uncompressed([%Message{key: key, value: value}|messages]) do
{message, msize} = create_message(value, key)
message_set = [<< 0 :: 64-signed >>, << msize :: 32-signed >>, message]
{message_set2, ms2size} = create_message_set(messages, :none)
{[message_set, message_set2], 8 + 4 + msize + ms2size}
end
+
+ defp create_message_set(messages, :none) do
+ create_message_set_uncompressed(messages)
+ end
defp create_message_set(messages, compression_type) do
{message_set, _} = create_message_set(messages, :none)
{compressed_message_set, attribute} =

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -289,7 +289,19 @@ defmodule KafkaEx.Server do
# credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
def kafka_server_produce(produce_request, state) do
correlation_id = state.correlation_id + 1
- produce_request_data = Produce.create_request(correlation_id, @client_id, produce_request)
+ produce_request_data = try do
+ Produce.create_request(correlation_id, @client_id, produce_request)
+ rescue
+ e in FunctionClauseError -> nil
+ end
+
+ case produce_request_data do
+ nil -> {:reply, {:error, "Invalid produce request"}, state}
+ _ -> kafka_server_produce_send_request(correlation_id, produce_request, produce_request_data, state)
+ end
+ end
+
+ def kafka_server_produce_send_request(correlation_id, produce_request, produce_request_data, state) do
{broker, state, corr_id} = case MetadataResponse.broker_for_topic(state.metadata, state.brokers, produce_request.topic, produce_request.partition) do
nil ->
{retrieved_corr_id, _} = retrieve_metadata(state.brokers, state.correlation_id, config_sync_timeout(), produce_request.topic)

MODIFY
test/integration/server0_p_8_p_0_test.exs
test/integration/server0_p_8_p_0_test.exs
@@ -43,6 +43,19 @@ defmodule KafkaEx.Server0P8P0.Test do
end)
end
+ @tag :producer_mem_leak
+ test "does not mem leak when message is not a Produce.Message" do
+ name = "create_topic_#{:rand.uniform(2000000)}"
+
+ assert {:error, "Invalid produce request"} == KafkaEx.produce(%KafkaEx.Protocol.Produce.Request{
+ messages: [%{key: "key1", value: "value1"}],
+ partition: 0,
+ required_acks: 0,
+ timeout: 500,
+ topic: name
+ })
+ end
+
test "when the partition is not found", %{worker: worker} do
partition = 42
assert :topic_not_found == KafkaEx.fetch(

Prevent memory leak in message producer when request is invalid