darnahsan
Broadway Kafka: failed to join group reason: {:client_down
I am trying to connect to upstash kafka using broadway and getting
[info] Group member (localhost-00,coor=#PID<0.531.0>,cb=#PID<0.528.0>,generation=0):
failed to join group
reason: {:client_down,
My broadway code is as below.
defmodule Maverick.Broadway.Upstash do
@moduledoc """
The Upstash Kafka context.
"""
use Broadway
@upstash_kafka_port Application.compile_env!(:broadway, :port)
@upstash_kafka_endpoint Application.compile_env!(:broadway, :endpoint)
@upstash_kafka_hosts [{@upstash_kafka_endpoint, @upstash_kafka_port}]
@upstash_kafka_authentication {Application.compile_env!(:broadway, :sasl_mechanism),
Application.compile_env!(:broadway, :username),
Application.compile_env!(:broadway, :password)}
@upstash_kafka_group_id "localhost-00"
@upstash_kafka_topic "localhost"
@producer_concurrency 1
@processors_concurrency 10
def start_link(_opts) do
IO.inspect(@upstash_kafka_authentication)
IO.inspect(@upstash_kafka_hosts)
Broadway.start_link(__MODULE__,
name: __MODULE__,
producer: [
module:
{BroadwayKafka.Producer,
[
hosts: @upstash_kafka_hosts,
group_id: @upstash_kafka_group_id,
topics: [@upstash_kafka_topic],
client_id_prefix: "localhost-maverick",
sasl: @upstash_kafka_authentication,
ssl: true
]},
concurrency: @producer_concurrency
],
processors: [
default: [
concurrency: @processors_concurrency
]
]
)
end
def handle_message(_, message, _) do
message |> IO.inspect()
message
end
end
Any help would be appreciated
Marked As Solved
darnahsan
Fixed it by updating the config as below, got some lead from How do I connect to Upstash Kafka with Broadway (ssl issues)
defmodule Maverick.Broadway.Upstash do
@moduledoc """
The Upstash Kafka context.
"""
use Broadway
@upstash_kafka_port Application.compile_env!(:broadway, :port)
@upstash_kafka_endpoint Application.compile_env!(:broadway, :endpoint)
@upstash_kafka_hosts [{@upstash_kafka_endpoint, @upstash_kafka_port}]
@upstash_kafka_authentication {Application.compile_env!(:broadway, :sasl_mechanism),
Application.compile_env!(:broadway, :username),
Application.compile_env!(:broadway, :password)}
@upstash_kafka_group_id "localhost-00"
@upstash_kafka_topic "localhost"
@producer_concurrency 1
@processors_concurrency 10
def start_link(_opts) do
IO.inspect(@upstash_kafka_authentication)
IO.inspect(@upstash_kafka_hosts)
Broadway.start_link(__MODULE__,
name: __MODULE__,
producer: [
module:
{BroadwayKafka.Producer,
[
hosts: @upstash_kafka_hosts,
group_id: @upstash_kafka_group_id,
topics: [@upstash_kafka_topic],
client_id_prefix: "localhost-maverick",
client_config: [
sasl: @upstash_kafka_authentication,
ssl: [
# from CAStore package
cacertfile: CAStore.file_path(),
verify_type: :verify_peer,
customize_hostname_check: [
match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
]
]
]
]},
concurrency: @producer_concurrency
],
processors: [
default: [
concurrency: @processors_concurrency
]
]
)
end
def handle_message(_, message, _) do
message |> IO.inspect()
message
end
end
1
Popular in Questions
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1]
15:22:35.803 [error] gen_event {lager_file_backend...
New
Hi all,
Trying to get some more clarity over utc_datetime and naive_datetime for Ecto:
https://hexdocs.pm/ecto/Ecto.Schema.html#module-...
New
How to set Jason to encode all fields in ecto schema, I don’t care about security and implementing only is taking long list of attributes...
New
i’m a new one to elixir
which editor can i use
vs code? or atom?
Thanks! :smiley:
New
Hi,
I need to transform a list of numbers into a map where the keys are the indexes and the values are the original values of the list....
New
To simplify some tasks at work, I wrote and published this package yesterday. It’s a simple macro that enables Access behaviour on struct...
New
Can someone explain the settings of pool_size of Ecto in config file? and what is the recommend size?
Thanks
New
Given a string, how can I get access to its character by index? Enum.at("my_string", 2) doesn't work.
Or rather, not char, but a substr...
New
In AR this is so simple
@articles = current_user.articles
How to do in Ecto?
def index(conn, _params) do
current_user = conn.assig...
New
Hi,
I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New
Other popular topics
Update:
How to use the Blogs & Podcasts section
You can post links to your blog posts or podcasts either in one of the Official Blog...
New
How to handle excepions in elixir? Suppose i have A, B, C ,D, E modules. and each module has get() function. A.get() method will call th...
New
Good day to you all.
I have been struggling to get a query involving like and ilike to work.
Can anyone assist me on this, please?
pro...
New
Hi, this is for people who, like me, have had some friction using .html.heex templates in VSCode.
The solution seems to be, in a hyphena...
New
I have an umbrella app.
Some of the apps inside depend on other apps in the umbrella, unsurprisingly.
I'm writing a test for one of t...
New
We have an ECS cluster with 4 services, where each task joins a single cluster, via discovery ECS discovery service.
Currently when I de...
New
I have a another noob question about loop. Since elixir is immutable, while loop is not directly possible.
total = 10
while total != 0
...
New
Hi all,
I've just started learning Elixir and Phoenix Framework, so please pardon my n00bness at this stage.
I'm trying to use Postg...
New
ExUnit now has describe blocks which is a welcome addition coming from RSpec. In the docs, it states that nested hierarchies of describe ...
New
Hello again - after a longish gap I’ve decided I really must dig into Elixir and see what’s been happening here - so I have a few questio...
New







