darnahsan
Broadway kafka producer not getting all partition assignment from Upstash Kafka
My understanding is that this is an upstash issue. I have a topic with 3 partitions but I only get assignment for 2 partitions. Below is my broadway code. Is there anything in the config that could be causing this missing partition assignment ?
defmodule Maverick.Broadway.Upstash do
@moduledoc """
The Upstash Kafka context.
"""
use Broadway
alias Maverick.Whatsapp.Message, as: WhatsappMessage
def upstash_kafka_port(), do: Application.fetch_env!(:maverick, :upstash_kafka_port)
def upstash_kafka_endpoint(), do: Application.fetch_env!(:maverick, :upstash_kafka_endpoint)
def upstash_kafka_hosts(), do: [{upstash_kafka_endpoint(), upstash_kafka_port()}]
def upstash_kafka_authentication(),
do:
{Application.fetch_env!(
:maverick,
:upstash_kafka_sasl_mechanism
), Application.fetch_env!(:maverick, :upstash_kafka_username),
Application.fetch_env!(:maverick, :upstash_kafka_password)}
def upstash_kafka_group_id(), do: "localhost-00"
def upstash_kafka_topic(), do: "localhost"
def producer_concurrency(), do: 2
def processors_concurrency(), do: 10
def start_link(_opts) do
Broadway.start_link(__MODULE__,
name: __MODULE__,
producer: [
module:
{BroadwayKafka.Producer,
[
hosts: upstash_kafka_hosts(),
group_id: upstash_kafka_group_id(),
topics: [upstash_kafka_topic()],
offset_reset_policy: :earliest,
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
First Post!
darnahsan
a new topic with 4 partitions got all partition assignments for now seem odd partition count is the problem, could the broadway config in any way be playing a part in this ?
0
Last Post!
mgibowski
Sorry, I was wrong in my previous message:
You should be fine with your original configuration, I don’t know what is the issue.
Regarding your other question - Erlang processes are lightweight processes and are not limited by the number of cores.
0
Popular in Questions
I have seen a lot of code which picks the first element from a list using Enum.at(0) instead of List.first. Is there a reason why people ...
New
Lets say I have map like this fetching from my database
%{"_id" => #BSON.ObjectId<58eb1a7a9ad169198c3dXXXX>, "email" => ...
New
I wanted to check elixir version in phoenix because i found that my elixir is 1.5 but when i use Enum.chunk_by it said the function is un...
New
In templates/appointment/index.html.eex:
<%= for appointment <- @appointments do %>
<tr>
<td><%= appoi...
New
I have a phoenix application from which a user can download multiple(5-6) files of size 1MB. I couldn’t find anything related to sending ...
New
As the title describes, I’m trying to run Enum.map() over a list of key/value pairs, where the value is a map. My data looks like this:
...
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
Other popular topics
I have seen a lot of code which picks the first element from a list using Enum.at(0) instead of List.first. Is there a reason why people ...
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 the...
New
Phoenix 1.4.0 released
Phoenix 1.4 is out! This release ships with exciting new features, most notably
with HTTP2 support, improved deve...
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
Hey,
Just curious what are the main benefits of Elixir compared to Clojure?
When is Elixir more useful than Clojure and vice versa?
Th...
New
After calling mix ecto.create I get this error:
17:00:32.162 [error] GenServer #PID<0.412.0> terminating
** (Postgrex.Error) FATAL...
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
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #hex












