Closed eddy147 closed 1 year ago
@eddy147 what API versions are you using?
I don't know which APIVersions kafka_ex uses, but this is my kafka_ex in mix.lock:
"kafka_ex": {:hex, :kafka_ex, "0.13.0", "2bfaf3c81d4ee01ed2088cb09e46c070c245f60f5752ec7043f29e807f6679ec", [:mix], [{:kayrock, "~> 0.1.12", [hex: :kayrock, repo: "hexpm", optional: false]}], "hexpm", "8a806eee5cd8191f45870b2ef4b3f4f52c57d798039f2d3fc602ce47053db7b9"},
@eddy147 Ok, looking at error you've pasted, sorry for not noticing it, you are using v0 everywhere.
api_versions: %{fetch: 0, offset_commit: 0, offset_fetch: 0}
I've tested my setup with 2.8.1, and it works. I'm using API version 3 & kayrock as a client without any issues. Could you try to bump it?
config :kafka_ex,
kafka_version: "kayrock",
api_versions: %{fetch: 3, offset_fetch: 3, offset_commit: 3}
will try thx!
Unfortunately, it didn't work.
I get the following error:
08:27:20.991 [debug] Successfully connected to broker "xx.x.xx.xxx:9092 (=> anonymized it)
08:27:20.994 [error] GenServer #PID<0.1331.0> terminating
** (CaseClauseError) no case clause matching: :unsupported_version
(kafka_ex 0.13.0) lib/kafka_ex/gen_consumer.ex:929: KafkaEx.GenConsumer.load_offsets/1
(kafka_ex 0.13.0) lib/kafka_ex/gen_consumer.ex:660: KafkaEx.GenConsumer.handle_info/2
(stdlib 4.2) gen_server.erl:1123: :gen_server.try_dispatch/4
(stdlib 4.2) gen_server.erl:1200: :gen_server.handle_msg/6
(stdlib 4.2) proc_lib.erl:240: :proc_lib.init_p_do_apply/3
Last message: :timeout
State: %KafkaEx.GenConsumer.State{consumer_module: Dockr.Kafka.Consumer, consumer_state: nil, commit_interval: 1000, commit_threshold: 100, worker_name: #PID<0.1332.0>, group: "dockr-cg", topic: "dockr", partition: 0, member_id: "kafka_ex-575f3dd6-deb8-432d-bea9-7d4ac0eb3adb", generation_id: 31, current_offset: nil, committed_offset: nil, acked_offset: nil, last_commit: nil, auto_offset_reset: :earliest, fetch_options: [auto_commit: false, worker_name: #PID<0.1332.0>], api_versions: %{fetch: 0, offset_commit: 0, offset_fetch: 0}}
08:27:20.996 [debug] Left consumer group dockr-cg
08:27:20.997 [notice] Application dockr exited: shutdown
{"Kernel pid terminated",application_controller,"{application_terminated,dockr,shutdown}"}
Kernel pid terminated (application_controller) ({application_terminated,dockr,shutdown})
Crash dump is being written to: erl_crash.dump...done
This is my complete config:
import Config
brokers =
if config_env() == :prod do
[{"xx.x.xx.xxx", 9092}] # anonymized it
else
[{"localhost", 9092}]
end
config :dockr, Dockr.Endpoint, port: 4000
config :dockr,
env: config_env(),
consumer_group_opts: [
heartbeat_interval: 1_000,
commit_interval: 1_000,
auto_offset_reset: :earliest
],
# 1st topic is from Lambda, which Dockr consumes, 2nd topic is from the Processor, which RTA consumes
kafka_topics: ["dockr", "aws-dockr"]
config :kafka_ex,
kafka_version: "kayrock",
api_versions: %{fetch: 3, offset_fetch: 3, offset_commit: 3},
brokers: brokers,
consumer_group: "dockr",
pool_size: 4,
use_ssl: false
I create a docker container from it and use that on the server. Is there maybe something wrong in my config? Locally, it works fine, if I use localhost as prod broker.
Not sure if it's not a config issue. I have a producer and a consumer, the producer is going OK, it's about starting the consumer that goes wrong, but I do not know what I do wrong, why kafka_ex would give me a :unsupported_version error:
defmodule Dockr.Kafka.ConsumerSupervisor do
@moduledoc """
This module has functions for starting the KafkaEx consumers. The caller is the supervisor coordinator.
"""
use DynamicSupervisor
alias Dockr.Kafka.Consumer
require Logger
def start_link(_) do
Logger.info("Starting #{__MODULE__}...")
DynamicSupervisor.start_link(__MODULE__, [], name: __MODULE__)
end
@impl true
def init(_) do
DynamicSupervisor.init(strategy: :one_for_one)
end
def start_worker do
topic = Application.get_env(:dockr, :kafka_topics) |> Enum.at(0)
child_spec = %{
id: KafkaEx.ConsumerGroup,
start: {
KafkaEx.ConsumerGroup,
:start_link,
[
Consumer,
"dockr-consumer-group",
[topic],
Application.get_env(:dockr, :consumer_group_opts)
]
}
}
DynamicSupervisor.start_child(__MODULE__, child_spec)
end
def stop_workers do
__MODULE__
|> Supervisor.which_children()
|> Enum.each(&DynamicSupervisor.terminate_child(__MODULE__, elem(&1, 1)))
end
end
And this is the supervisor that starts the ConsumerSupervisor:
defmodule Dockr.Kafka.SupervisorCoordinator do
@moduledoc """
This module is responsible for starting the KafkaEx application and the producers and consumers.
It also checks periodically if the KafkaEx application is still running and restarts it if necessary.
"""
use GenServer
require Logger
alias Dockr.Kafka.ConsumerSupervisor
alias Dockr.Kafka.ProducerSupervisor
@interval 30_000
def start_link(_) do
Logger.info("Starting #{__MODULE__}...")
GenServer.start_link(__MODULE__, [], name: __MODULE__)
end
def init([]) do
schedule_next()
{:ok, [], {:continue, :start_workers}}
end
def handle_continue(:start_workers, state) do
start_consumer_supervisor()
start_consumers()
{:noreply, state}
end
def handle_info(:tick, state) do
if kafka_down?() do
start_kafka()
end
if consumer_supervisor_down?() do
start_consumer_supervisor()
end
if any_consumer_down?() do
start_consumers()
end
schedule_next()
{:noreply, state}
end
defp kafka_down? do
Process.whereis(:kafka_ex) == nil
end
defp start_kafka do
Logger.warn("KafkaEx application is not running, starting it...")
Application.start(:kafka_ex)
end
defp consumer_supervisor_down? do
Process.whereis(ConsumerSupervisor) == nil
end
defp start_consumer_supervisor() do
Supervisor.start_child(
Dockr.Application,
%{
id: ConsumerSupervisor,
start: {ConsumerSupervisor, :start_link, [[]]},
type: :supervisor,
restart: :temporary
}
)
end
defp any_consumer_down?() do
Supervisor.count_children(ConsumerSupervisor).active < 1
end
defp start_consumers() do
ConsumerSupervisor.stop_workers()
ConsumerSupervisor.start_worker()
end
defp schedule_next do
Process.send_after(self(), :tick, @interval)
end
end
For anyone who faces the same issue: the reson for this was that the Kafka that was installed was Confluent Kafka, which apparantly uses a different protocol which caused the issue.
I would have liked that the error was a bit more verbose than :unsupported_version.
On our staging server we have Kafka 2.8.1 installed. Running our app with kafka_ex 0.13 gives a :unsupported_version error: (our app is called dockr, not a typo )
Installed the same version of Kafka (2.8.0) locally and it all works well locally.
What can cause this error?
Many thx in advance.