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
If I have a post route which an argument:
post /my_post_route/:my_param1, MyController.my_post_handler
How would get the post params ...
New
can someone please explain to me how Enum.reduce works with maps
New
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
I am trying to figure out how Mix knows whether the environment is test, dev, or prod – where is this set?
Thanks.
New
How to bind a phoenix app to a specific ip address?
could not find anything about that, nowhere, unfortunately, but for me this is quite...
New
lets say i have a sample like
a = 20; b = 10;
if (a > b) do
{:ok, "a"}
end
if (a < b) do
{:ok, b}
end
if (a == b) do
{:ok, "equa...
New
I’m brand new to Phoenix and I have stripped one of the demo applications to the bone. I just want to get an svg up on the screen. Here i...
New
I will often find my self writing things similar to:
case some_value do
nil -> something()
"" -> something()
_ -> somethi...
New
Hi! May someone helps me, please!
I have two apps into an umbrella project: the first one is Database, which manages queries, and the se...
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 t...
New
Other popular topics
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
Hello all!
I am typing this post from my new MacBook Pro with the M1 chip. I’m loving it so far, and will probably use it as my daily dr...
New
I am trying to figure out how Mix knows whether the environment is test, dev, or prod – where is this set?
Thanks.
New
This release brings a number of exciting features, including integration with the new Phoenix LiveDashboard and Phoenix LiveView. There h...
New
I have a server on AWS, and was running a load test using artillery. When looking at the Phoenix dashboard I see the Ports going to 100% ...
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
i’m a new one to elixir
which editor can i use
vs code? or atom?
Thanks! :smiley:
New
I asked this very same question on twitter and got some interesting feedback, but I thought it would be a good question to ask here as we...
New
What is the proper way to load a module from a file in to IEX?
In the python world, doing something like this pretty standard:
from ....
New
Hi everyone!
I need implement if…else if…else condition from my elixir code, and anymore of this control flow structures not work proper...
New
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #advent-of-code
- #elixirconf-us
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #performance









