Memory leak when using KafkaEx.stream

[dplummer]

I mentioned on slack:

I’ve written an app (https://github.com/avvo/kafkamon) that consumes kafka using KafkaEx. I seem to have a memory leak using KafkaEx.stream (https://github.com/avvo/kafkamon/blob/master/lib/reader/event_queue/consumer.ex#L25). Has anyone used that api and seen a similar issue?

And @dantswain suggested I create an issue after reproducing it with a minimal app. I built KafkaConsumerTest and was able to reproduce the behavior I saw reliably. The README in the project outlines how to run the consumer and producer.

I reproduced this with:

Erlang/OTP 19 [erts-8.0.2] [source] [64-bit] [smp:8:8] [async-threads:10] [hipe] [kernel-poll:false] [dtrace]

Elixir 1.3.2

And both v0.5.0 and github master of KafkaEx. My example producer creates a random binary, but the same behavior is observed for a string.

Observed behavior on my MacBook Pro

Ran the producer twice and captured the memory and process graphs from the observer. Memory did not decrease after 5 minutes of no activity, except in Elixir.KafkaEx.Server, which did decrease to ~143k (from 264k).

[dplummer]

I tried using KafkaEx.fetch and did not observer a memory leak:

  def init(:ok) do
    GenServer.cast self, :fetch
    {:ok, latest_offset(@topic)}
  end

  def handle_cast(:fetch, offset) do
    Logger.info "Fetch of #{@topic}##{offset}"

    [%{partitions: [%{message_set: messages}]}] = KafkaEx.fetch(@topic, 0, offset: offset)

    new_offset = messages
    |> Enum.reduce(0, fn message, max ->
      current_offset = process_message(message)
      cond do
        current_offset > max -> current_offset
        true -> max
      end
    end)

    Logger.info "Done with our fetch, new_offset is #{new_offset}"
    Process.sleep 1000
    GenServer.cast self, :fetch
    {:noreply, new_offset}
  end

This is after processing 250k messages:

[dplummer]

Forgot to mention, am running kafka 0.8.2.1

[dplummer]

So I think the issue is the GenEvent handler on server.ex#L319. I'm pretty sure that's the "leak". I'm not using the handler, just basically using GenEvent.stream(event_stream) |> Stream.each(&stuff/1) |> Stream.run.

It appears I could use the handler by calling something like GenEvent.call(stream.manager, KafkaEx.Handler, :messages) |> Enum.take(2). But I think I should have the option of not using a handler.

What do you think of something like KafkaEx.stream("x", 0, handler: nil) to skip setting up the handler?

Alternatively, I could define a null handler like:

defmodule NullHandler do
  use GenEvent

  def handle_event(_, _), do: {:ok, []}

  def handle_call(:messages, _), do: {:ok, [], []}
end

Using the NullHandler in my test app fixes it:

Maybe make the NullHandler the default?

[dantswain]

Great detective work @dplummer!

We'll take a closer look at this ASAP. In the meantime, it seems like you have a workaround?

[dplummer]

Yea, adding handler: NullHandler to the stream is a good workaround for me.

[dantswain]

@dplummer The stream rewrite is complete. This should no longer be an issue. I'm going to close this issue because we were pretty confident that the implementation using GenEvent was causing the leak and the new implementation doesn't use GenEvent at all. If there is a memory leak with the new implementation, it should be a new issue as it almost certainly has a different underlying cause.

Thanks again for your detective work on this!