ConsumerGroup can accept alternative Consumer implementations
MODIFY
.gitignore
.gitignore
@@ -8,3 +8,5 @@ kafka1
kafka2
kafka3
doc
+
+.elixir_ls
\ No newline at end of file

MODIFY
lib/kafka_ex/consumer_group.ex
lib/kafka_ex/consumer_group.ex
@@ -137,12 +137,32 @@ defmodule KafkaEx.ConsumerGroup do
"""
@spec start_link(module, binary, [binary], options) :: Supervisor.on_start()
def start_link(consumer_module, group_name, topics, opts \\ []) do
+ start_link(KafkaEx.GenConsumer, consumer_module, group_name, topics, opts)
+ end
+
+ @doc """
+ Starts a consumer group process tree with an alternate consumer implementation.
+
+ By default starts `KafkaEx.GenConsumer` But can be passed any module
+ implementing `GenServer`. The consumer is responsible for pulling messages and
+ managing its commit state. For example `KafkExGenStageConsumer` implements a
+ `GenStage` producer which pulls messages from kafka according to demand.
+ """
+ @spec start_link(module, module, binary, [binary], options) ::
+ Supervisor.on_start()
+ def start_link(
+ gen_consumer_module,
+ consumer_module,
+ group_name,
+ topics,
+ opts
+ ) do
{supervisor_opts, module_opts} =
Keyword.split(opts, [:name, :strategy, :max_restarts, :max_seconds])
Supervisor.start_link(
__MODULE__,
- {consumer_module, group_name, topics, module_opts},
+ {gen_consumer_module, consumer_module, group_name, topics, module_opts},
supervisor_opts
)
end
@@ -284,11 +304,18 @@ defmodule KafkaEx.ConsumerGroup do
# used by ConsumerGroup.Manager to set partition assignments
@doc false
- def start_consumer(pid, consumer_module, group_name, assignments, opts) do
+ def start_consumer(
+ pid,
+ gen_consumer_module,
+ consumer_module,
+ group_name,
+ assignments,
+ opts
+ ) do
child =
supervisor(
KafkaEx.GenConsumer.Supervisor,
- [consumer_module, group_name, assignments, opts],
+ [gen_consumer_module, consumer_module, group_name, assignments, opts],
id: :consumer
)
@@ -311,13 +338,13 @@ defmodule KafkaEx.ConsumerGroup do
end
@doc false
- def init({consumer_module, group_name, topics, opts}) do
+ def init({gen_consumer_module, consumer_module, group_name, topics, opts}) do
opts = Keyword.put(opts, :supervisor_pid, self())
children = [
worker(
KafkaEx.ConsumerGroup.Manager,
- [consumer_module, group_name, topics, opts]
+ [gen_consumer_module, consumer_module, group_name, topics, opts]
)
]

MODIFY
lib/kafka_ex/consumer_group/manager.ex
lib/kafka_ex/consumer_group/manager.ex
@@ -23,6 +23,7 @@ defmodule KafkaEx.ConsumerGroup.Manager do
:worker_name,
:heartbeat_interval,
:session_timeout,
+ :gen_consumer_module,
:consumer_module,
:consumer_opts,
:partition_assignment_callback,
@@ -50,22 +51,34 @@ defmodule KafkaEx.ConsumerGroup.Manager do
@doc false
# use `KafkaEx.ConsumerGroup.start_link/4` instead
- @spec start_link(module, binary, [binary], KafkaEx.GenConsumer.options()) ::
+ @spec start_link(
+ module,
+ module,
+ binary,
+ [binary],
+ KafkaEx.GenConsumer.options()
+ ) ::
GenServer.on_start()
- def start_link(consumer_module, group_name, topics, opts \\ []) do
+ def start_link(
+ gen_consumer_module,
+ consumer_module,
+ group_name,
+ topics,
+ opts \\ []
+ ) do
gen_server_opts = Keyword.get(opts, :gen_server_opts, [])
consumer_opts = Keyword.drop(opts, [:gen_server_opts])
GenServer.start_link(
__MODULE__,
- {consumer_module, group_name, topics, consumer_opts},
+ {gen_consumer_module, consumer_module, group_name, topics, consumer_opts},
gen_server_opts
)
end
# GenServer callbacks
- def init({consumer_module, group_name, topics, opts}) do
+ def init({gen_consumer_module, consumer_module, group_name, topics, opts}) do
heartbeat_interval =
Keyword.get(
opts,
@@ -114,6 +127,7 @@ defmodule KafkaEx.ConsumerGroup.Manager do
heartbeat_interval: heartbeat_interval,
session_timeout: session_timeout,
consumer_module: consumer_module,
+ gen_consumer_module: gen_consumer_module,
partition_assignment_callback: partition_assignment_callback,
consumer_opts: consumer_opts,
group_name: group_name,
@@ -380,6 +394,7 @@ defmodule KafkaEx.ConsumerGroup.Manager do
defp start_consumer(
%State{
consumer_module: consumer_module,
+ gen_consumer_module: gen_consumer_module,
consumer_opts: consumer_opts,
group_name: group_name,
supervisor_pid: pid
@@ -389,6 +404,7 @@ defmodule KafkaEx.ConsumerGroup.Manager do
{:ok, consumer_supervisor_pid} =
ConsumerGroup.start_consumer(
pid,
+ gen_consumer_module,
consumer_module,
group_name,
assignments,

MODIFY
lib/kafka_ex/gen_consumer.ex
lib/kafka_ex/gen_consumer.ex
@@ -175,7 +175,7 @@ defmodule KafkaEx.GenConsumer do
@partition 0
setup do
- {:ok, state} = ExampleGenConsumer.init(@topic, @partition)
+ {k:ok, state} = ExampleGenConsumer.init(@topic, @partition)
{:ok, %{state: state}}
end

MODIFY
lib/kafka_ex/gen_consumer/supervisor.ex
lib/kafka_ex/gen_consumer/supervisor.ex
@@ -38,11 +38,11 @@ defmodule KafkaEx.GenConsumer.Supervisor do
],
KafkaEx.GenConsumer.options()
) :: Elixir.Supervisor.on_start()
- def start_link(consumer_module, group_name, assignments, opts \\ []) do
+ def start_link(gen_consumer_module, consumer_module, group_name, assignments, opts \\ []) do
start_link_result =
Elixir.Supervisor.start_link(
__MODULE__,
- {consumer_module, group_name, assignments, opts}
+ {gen_consumer_module, consumer_module, group_name, assignments, opts}
)
case start_link_result do
@@ -77,9 +77,9 @@ defmodule KafkaEx.GenConsumer.Supervisor do
|> Enum.any?(&Process.alive?/1)
end
- def init({consumer_module, group_name, _assignments, _opts}) do
+ def init({gen_consumer_module, consumer_module, group_name, _assignments, _opts}) do
children = [
- worker(KafkaEx.GenConsumer, [consumer_module, group_name])
+ worker(gen_consumer_module, [consumer_module, group_name])
]
supervise(children, strategy: :simple_one_for_one)

MODIFY
mix.lock
mix.lock
"certifi": {:hex, :certifi, "1.2.1", "c3904f192bd5284e5b13f20db3ceac9626e14eeacfbb492e19583cf0e37b22be", [:rebar3], [], "hexpm"},
"credo": {:hex, :credo, "0.8.10", "261862bb7363247762e1063713bb85df2bbd84af8d8610d1272cd9c1943bba63", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}], "hexpm"},
"dialyxir": {:hex, :dialyxir, "1.0.0-rc.3", "774306f84973fc3f1e2e8743eeaa5f5d29b117f3916e5de74c075c02f1b8ef55", [:mix], [], "hexpm"},
- "dialyze": {:hex, :dialyze, "0.2.1", "9fb71767f96649020d769db7cbd7290059daff23707d6e851e206b1fdfa92f9d", [:mix], []},
"earmark": {:hex, :earmark, "1.2.4", "99b637c62a4d65a20a9fb674b8cffb8baa771c04605a80c911c4418c69b75439", [:mix], [], "hexpm"},
"ex_doc": {:hex, :ex_doc, "0.18.3", "f4b0e4a2ec6f333dccf761838a4b253d75e11f714b85ae271c9ae361367897b7", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
"excoveralls": {:hex, :excoveralls, "0.7.1", "3dd659db19c290692b5e2c4a2365ae6d4488091a1ba58f62dcbdaa0c03da5491", [:mix], [{:exjsx, ">= 3.0.0", [hex: :exjsx, repo: "hexpm", optional: false]}, {:hackney, ">= 0.12.0", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm"},
"exjsx": {:hex, :exjsx, "4.0.0", "60548841e0212df401e38e63c0078ec57b33e7ea49b032c796ccad8cde794b5c", [:mix], [{:jsx, "~> 2.8.0", [hex: :jsx, repo: "hexpm", optional: false]}], "hexpm"},
- "fs": {:hex, :fs, "0.9.2"},
"hackney": {:hex, :hackney, "1.8.6", "21a725db3569b3fb11a6af17d5c5f654052ce9624219f1317e8639183de4a423", [:rebar3], [{:certifi, "1.2.1", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "5.0.2", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "1.0.2", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.1", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm"},
"idna": {:hex, :idna, "5.0.2", "ac203208ada855d95dc591a764b6e87259cb0e2a364218f215ad662daa8cd6b4", [:rebar3], [{:unicode_util_compat, "0.2.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm"},
"jsx": {:hex, :jsx, "2.8.2", "7acc7d785b5abe8a6e9adbde926a24e481f29956dd8b4df49e3e4e7bcc92a018", [:mix, :rebar3], [], "hexpm"},
"metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm"},
"mimerl": {:hex, :mimerl, "1.0.2", "993f9b0e084083405ed8252b99460c4f0563e41729ab42d9074fd5e52439be88", [:rebar3], [], "hexpm"},
- "mix_test_watch": {:hex, :mix_test_watch, "0.2.6"},
"snappy": {:git, "https://github.com/fdmanana/snappy-erlang-nif", "c4cd1bb35f3a3a399737d64bd03b479817b90e75", []},
"ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.1", "28a4d65b7f59893bc2c7de786dec1e1555bd742d336043fe644ae956c3497fbe", [:make, :rebar], [], "hexpm"},
"unicode_util_compat": {:hex, :unicode_util_compat, "0.2.0", "dbbccf6781821b1c0701845eaf966c9b6d83d7c3bfc65ca2b78b88b8678bfa35", [:rebar3], [], "hexpm"},

ConsumerGroup can accept alternative Consumer implementations