Delete the dead stream code.
DELETE
lib/kafka_ex/handler.ex
-defmodule KafkaEx.Handler do
- @moduledoc """
- Default GenEvent handler for KafkaEx.stream
-
- Received message sets are accumulated in the GenEvent state and can
- be retrieved and flushed via the `:messages` call.
- """
-
- use GenEvent
-
- def handle_event(message_set, message_sets) do
- {:ok, [message_set|message_sets]}
- end
-
- def handle_call(:messages, message_sets) do
- {:ok, Enum.reverse(message_sets), []}
- end
-end

MODIFY
lib/kafka_ex/server.ex
lib/kafka_ex/server.ex
@@ -21,7 +21,6 @@ defmodule KafkaEx.Server do
metadata: %Metadata.Response{},
brokers: [],
event_pid: nil,
- stream_timer: nil,
consumer_metadata: %ConsumerMetadata.Response{},
correlation_id: 0,
consumer_group: nil,
@@ -36,7 +35,6 @@ defmodule KafkaEx.Server do
metadata: Metadata.Response.t,
brokers: [Broker.t],
event_pid: nil | pid,
- stream_timer: reference,
consumer_metadata: ConsumerMetadata.Response.t,
correlation_id: integer,
metadata_update_interval: nil | integer,
@@ -136,21 +134,6 @@ defmodule KafkaEx.Server do
{:noreply, new_state, timeout | :hibernate} |
{:stop, reason, reply, new_state} |
{:stop, reason, new_state} when reply: term, new_state: term, reason: term
- @callback kafka_server_create_stream(handler :: term, handler_init :: term, state :: State.t) ::
- {:reply, reply, new_state} |
- {:reply, reply, new_state, timeout | :hibernate} |
- {:noreply, new_state} |
- {:noreply, new_state, timeout | :hibernate} |
- {:stop, reason, reply, new_state} |
- {:stop, reason, new_state} when reply: term, new_state: term, reason: term
- @callback kafka_server_start_streaming(fetch_request :: FetchRequest.t, state :: State.t) ::
- {:noreply, new_state} |
- {:noreply, new_state, timeout | :hibernate} |
- {:stop, reason :: term, new_state} when new_state: term
- @callback kafka_server_stop_streaming(state :: State.t) ::
- {:noreply, new_state} |
- {:noreply, new_state, timeout | :hibernate} |
- {:stop, reason :: term, new_state} when new_state: term
@callback kafka_server_update_metadata(state :: State.t) ::
{:noreply, new_state} |
{:noreply, new_state, timeout | :hibernate} |
@@ -253,24 +236,6 @@ defmodule KafkaEx.Server do
kafka_server_heartbeat(group_name, generation_id, member_id, state)
end
- def handle_call({:create_stream, handler, handler_init}, _from, state) do
- kafka_server_create_stream(handler, handler_init, state)
- end
-
- def handle_info({:start_streaming, fetch_request}, state) do
- kafka_server_start_streaming(fetch_request, state)
- end
-
- def handle_info(:stop_streaming, state) do
- {:noreply, state} = kafka_server_stop_streaming(state)
- state = case state.stream_timer do
- nil -> state
- ref -> Process.cancel_timer(ref)
- %{state | stream_timer: nil}
- end
- {:noreply, state}
- end
-
def handle_info(:update_metadata, state) do
kafka_server_update_metadata(state)
end
@@ -360,25 +325,6 @@ defmodule KafkaEx.Server do
{:reply, metadata, updated_state}
end
- def kafka_server_create_stream(handler, handler_init, state) do
- new_state = if state.event_pid && Process.alive?(state.event_pid) do
- Logger.log(:warn, "'#{state.worker_name}' already streaming handler '#{handler}'")
- state
- else
- {:ok, event_pid} = GenEvent.start_link
- updated_state = %{state | event_pid: event_pid}
- :ok = GenEvent.add_handler(updated_state.event_pid, handler, handler_init)
- updated_state
- end
- {:reply, GenEvent.stream(new_state.event_pid), new_state}
- end
-
- def kafka_server_stop_streaming(state) do
- Logger.log(:debug, "Stopped worker #{inspect state.worker_name} from streaming")
- GenEvent.stop(state.event_pid)
- {:noreply, %{state | event_pid: nil}}
- end
-
def kafka_server_update_metadata(state) do
{:noreply, update_metadata(state)}
end
@@ -419,8 +365,7 @@ defmodule KafkaEx.Server do
defoverridable [
kafka_server_produce: 2, kafka_server_offset: 4,
- kafka_server_metadata: 2, kafka_server_create_stream: 3,
- kafka_server_stop_streaming: 1, kafka_server_update_metadata: 1,
+ kafka_server_metadata: 2, kafka_server_update_metadata: 1,
]
defp remove_stale_brokers(brokers, metadata_brokers) do

MODIFY
lib/kafka_ex/server_0_p_8_p_0.ex
lib/kafka_ex/server_0_p_8_p_0.ex
@@ -67,31 +67,6 @@ defmodule KafkaEx.Server0P8P0 do
def kafka_server_heartbeat(_, _, _, _state), do: raise "Heartbeat is not supported in 0.8.0 version of kafka"
def kafka_server_update_consumer_metadata(_state), do: raise "Consumer Group Metadata is not supported in 0.8.0 version of kafka"
- def kafka_server_start_streaming(_, state = %State{event_pid: nil}) do
- # our streaming could have been canceled with a streaming update in-flight
- {:noreply, state}
- end
- def kafka_server_start_streaming(fetch_request, state) do
- {response, state} = fetch(fetch_request, state)
- offset = case response do
- :topic_not_found ->
- fetch_request.offset
- _ ->
- message = response |> hd |> Map.get(:partitions) |> hd
- Enum.each(message.message_set, fn(message_set) -> GenEvent.notify(state.event_pid, message_set) end)
- case message.last_offset do
- nil -> fetch_request.offset
- last_offset -> last_offset + 1
- end
- end
-
- ref = Process.send_after(
- self(), {:start_streaming, %{fetch_request | offset: offset}}, 500
- )
-
- {:noreply, %{state | stream_timer: ref}}
- end
-
defp fetch(fetch_request, state) do
fetch_data = Fetch.create_request(%FetchRequest{
fetch_request |

MODIFY
lib/kafka_ex/server_0_p_8_p_2.ex
lib/kafka_ex/server_0_p_8_p_2.ex
@@ -111,33 +111,6 @@ defmodule KafkaEx.Server0P8P2 do
{:reply, consumer_metadata, state}
end
- def kafka_server_start_streaming(_, state = %State{event_pid: nil}) do
- # our streaming could have been canceled with a streaming update in-flight
- {:noreply, state}
- end
- def kafka_server_start_streaming(fetch_request, state) do
- true = consumer_group_if_auto_commit?(fetch_request.auto_commit, state)
-
- {response, state} = fetch(fetch_request, state)
- offset = case response do
- :topic_not_found ->
- fetch_request.offset
- _ ->
- message = response |> hd |> Map.get(:partitions) |> hd
- Enum.each(message.message_set, fn(message_set) -> GenEvent.notify(state.event_pid, message_set) end)
- case message.last_offset do
- nil -> fetch_request.offset
- last_offset -> last_offset + 1
- end
- end
-
- ref = Process.send_after(
- self(), {:start_streaming, %{fetch_request | offset: offset}}, 500
- )
-
- {:noreply, %{state | stream_timer: ref}}
- end
-
def kafka_server_update_consumer_metadata(state) do
true = consumer_group?(state)
{_, state} = update_consumer_metadata(state)

MODIFY
lib/kafka_ex/server_0_p_9_p_0.ex
lib/kafka_ex/server_0_p_9_p_0.ex
@@ -34,7 +34,6 @@ defmodule KafkaEx.Server0P9P0 do
defdelegate kafka_server_offset_fetch(offset_fetch, state), to: Server0P8P2
defdelegate kafka_server_offset_commit(offset_commit_request, state), to: Server0P8P2
defdelegate kafka_server_consumer_group_metadata(state), to: Server0P8P2
- defdelegate kafka_server_start_streaming(fetch_request, state), to: Server0P8P2
defdelegate kafka_server_update_consumer_metadata(state), to: Server0P8P2
def kafka_server_init([args]) do

Delete the dead stream code.