Wrap gen_consumer_module and consumer_module in a tuple to prevent linter complaints about arity
MODIFY
lib/kafka_ex/consumer_group.ex
lib/kafka_ex/consumer_group.ex
@@ -137,7 +137,7 @@ 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)
+ start_link({KafkaEx.GenConsumer, consumer_module}, group_name, topics, opts)
end
@doc """
@@ -148,11 +148,10 @@ defmodule KafkaEx.ConsumerGroup do
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) ::
+ @spec start_link({module, module}, binary, [binary], options) ::
Supervisor.on_start()
def start_link(
- gen_consumer_module,
- consumer_module,
+ {gen_consumer_module, consumer_module},
group_name,
topics,
opts
@@ -162,7 +161,7 @@ defmodule KafkaEx.ConsumerGroup do
Supervisor.start_link(
__MODULE__,
- {gen_consumer_module, consumer_module, group_name, topics, module_opts},
+ {{gen_consumer_module, consumer_module}, group_name, topics, module_opts},
supervisor_opts
)
end
@@ -306,8 +305,7 @@ defmodule KafkaEx.ConsumerGroup do
@doc false
def start_consumer(
pid,
- gen_consumer_module,
- consumer_module,
+ {gen_consumer_module, consumer_module},
group_name,
assignments,
opts
@@ -315,7 +313,7 @@ defmodule KafkaEx.ConsumerGroup do
child =
supervisor(
KafkaEx.GenConsumer.Supervisor,
- [gen_consumer_module, consumer_module, group_name, assignments, opts],
+ [{gen_consumer_module, consumer_module}, group_name, assignments, opts],
id: :consumer
)
@@ -338,13 +336,13 @@ defmodule KafkaEx.ConsumerGroup do
end
@doc false
- def init({gen_consumer_module, 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,
- [gen_consumer_module, 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
@@ -52,8 +52,7 @@ defmodule KafkaEx.ConsumerGroup.Manager do
@doc false
# use `KafkaEx.ConsumerGroup.start_link/4` instead
@spec start_link(
- module,
- module,
+ {module, module},
binary,
[binary],
KafkaEx.GenConsumer.options()
@@ -71,14 +70,15 @@ defmodule KafkaEx.ConsumerGroup.Manager do
GenServer.start_link(
__MODULE__,
- {gen_consumer_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({gen_consumer_module, consumer_module, group_name, topics, opts}) do
+ def init({{gen_consumer_module, consumer_module}, group_name, topics, opts}) do
heartbeat_interval =
Keyword.get(
opts,
@@ -404,8 +404,7 @@ defmodule KafkaEx.ConsumerGroup.Manager do
{:ok, consumer_supervisor_pid} =
ConsumerGroup.start_consumer(
pid,
- gen_consumer_module,
- consumer_module,
+ {gen_consumer_module, consumer_module},
group_name,
assignments,
consumer_opts

MODIFY
lib/kafka_ex/gen_consumer/supervisor.ex
lib/kafka_ex/gen_consumer/supervisor.ex
@@ -31,18 +31,22 @@ defmodule KafkaEx.GenConsumer.Supervisor do
returns `{:ok, pid}`, where `pid` is the PID of the supervisor.
"""
@spec start_link(
- callback_module :: module,
- consumer_group_name :: binary,
+ {callback_module :: module, consumer_group_name :: binary},
assigned_partitions :: [
{topic_name :: binary, partition_id :: non_neg_integer}
],
KafkaEx.GenConsumer.options()
) :: Elixir.Supervisor.on_start()
- def start_link(gen_consumer_module, 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__,
- {gen_consumer_module, consumer_module, group_name, assignments, opts}
+ {{gen_consumer_module, consumer_module}, group_name, assignments, opts}
)
case start_link_result do
@@ -77,7 +81,10 @@ defmodule KafkaEx.GenConsumer.Supervisor do
|> Enum.any?(&Process.alive?/1)
end
- def init({gen_consumer_module, consumer_module, group_name, _assignments, _opts}) do
+ def init(
+ {{gen_consumer_module, consumer_module}, group_name, _assignments,
+ _opts}
+ ) do
children = [
worker(gen_consumer_module, [consumer_module, group_name])
]

Wrap gen_consumer_module and consumer_module in a tuple to prevent linter complaints about arity