I am using an external application to authorize requests before sending to
broker. Before kraft I initialized aclauthroizer and just called configure and
then called authorize (the zk watcher inside aclauthorizer took care of loading
and updating acls).
In case of kraft, how to use this
bels%20%3D%20newbie
Sanskar Jhajharia
Sr. Software Engineer @ Confluent
Contact: Personal | Official
On Wed, Jun 11, 2025 at 7:00 PM Vinothini A wrote:
> Hi team,
> I want to contribute to kafka.
> Im a newbie may i know how to start and test the application.
>
> can you plea
Hi team,
I want to contribute to kafka.
Im a newbie may i know how to start and test the application.
can you please help here.
Regards,
vinothini.
Hello,
We have a Kafka Streams consumer application running Kafka Streams 3.4,
with our Kafka brokers running 2.6. This consumer application consumes from
two topics with 250 partitions each, and we co-partition them to ensure
each task is consuming from the same partitions in each topic. Some
a and larger linger.ms settings.
However, like you, I've found that a single KafkaProducer instance can
sometimes become a bottleneck. For instance, threads can be blocked in
certain cases when they are producing to the same partition. In our
application, we found a producer per thread wasn't
Santhosh Gopal
On Mon, Apr 24, 2023 at 9:01 AM Arvid Sundbom
wrote:
Hi!
I have a question about executing a Kafka producer application,
utilizing
multiple threads.
In the documentation for Kafka producers (
https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/producer
wrote:
> Hi!
> I have a question about executing a Kafka producer application, utilizing
> multiple threads.
> In the documentation for Kafka producers (
>
> https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html
> )
> it says t
e you, I've found that a single KafkaProducer instance can
sometimes become a bottleneck. For instance, threads can be blocked in certain
cases when they are producing to the same partition. In our application, we
found a producer per thread wasn't necessary, but adding a small pool of
Hi!
I have a question about executing a Kafka producer application, utilizing
multiple threads.
In the documentation for Kafka producers (
https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html)
it says that "...sharing a single producer instance a
additional API implementations in order to
support EOS.
Guozhang
On Mon, Apr 3, 2023 at 5:06 AM Pushkar Deole wrote:
>
> Hi All,
>
> We are using streams application with redis for state store.
> Redis was mainly considered instead of kafka state stores because of the
> reason that globa
Hi All,
We are using streams application with redis for state store.
Redis was mainly considered instead of kafka state stores because of the
reason that global state store once updated by one application instance was
taking few milliseconds to reflect updated global state to another
application
g to track down an issue where the stream processors are
> not correctly configured when the application starts up before Kafka is up.
> Specifically, all of the processor nodes are correctly created except for
> the KTABLE-SINK-0# node. The result of this is that the services
&g
where the stream processors are
not correctly configured when the application starts up before Kafka is up.
Specifically, all of the processor nodes are correctly created except for
the KTABLE-SINK-0# node. The result of this is that the services
consume messages but do not forward their
t; Otherwise, I think the best path forward would be to file a bug report
> > > on the Apache Kafka Jira with enough information to reproduce the issue
> > > (or if you’re able to provide a repro, that would be awesome).
> > >
> > > Thanks, and sorr
f you’re able to provide a repro, that would be awesome).
> >
> > Thanks, and sorry for the trouble.
> > -John
> >
> > On Tue, Sep 27, 2022, at 03:15, Tomasz Gac wrote:
> >> I upgraded to kafka streams 3.0.0 with positive task.max.idle.ms and
> it d
fka streams 3.0.0 with positive task.max.idle.ms and it did
>> not help.
>> When lag is large, the application still consumes data batches without
>> interleaving.
>>
>>
>>
>> wt., 27 wrz 2022 o 05:51 John Roesler napisał(a):
>>
>>> Hi T
ith positive task.max.idle.ms and it did
> not help.
> When lag is large, the application still consumes data batches without
> interleaving.
>
>
>
> wt., 27 wrz 2022 o 05:51 John Roesler napisał(a):
>
>> Hi Tomasz,
>>
>> Thanks for asking. This soun
I upgraded to kafka streams 3.0.0 with positive task.max.idle.ms and it did
not help.
When lag is large, the application still consumes data batches without
interleaving.
wt., 27 wrz 2022 o 05:51 John Roesler napisał(a):
> Hi Tomasz,
>
> Thanks for asking. This sounds like the situa
?
Thanks,
-John
On Mon, Sep 26, 2022, at 01:35, Tomasz Gac wrote:
> Hi group,
>
> I wrote a simple kafka streams application with topology such as below:
>
> builder.addStateStore(
>> Stores.keyValueStoreBuilder(
>> Stores.persistentKeyValueStore(&qu
Hi group,
I wrote a simple kafka streams application with topology such as below:
builder.addStateStore(
> Stores.keyValueStoreBuilder(
> Stores.persistentKeyValueStore("STORE"),
> Serdes.String(), Serdes.String())
> .withLoggingEnabled(storeConfi
Hello
The idea behind a GlobalKTable is to materiliaze data (a kafka topic)
close from where it is used. Actually, each task/worker will materialize
the full GlobalKTable in order to use it. So in your scenario, what
should be shared between your services is ideally the Kafka topic used
to bu
we have 2 microservices:
responsible for consuming message and update a globalKtable with
configuration information (key=id, value=myObject)
once it is activated with some id as an input it should look into the
configuration globalKtable and retrieve data by that id
how is it possible to access
sure that yoru write will succeed, before the next
commit may happen.
Yes, there can be multiple sink nodes.
-Matthias
On 7/14/21 10:09 PM, Pushkar Deole wrote:
> That's exactly what my question was: since there is an external producer in
> the application without a sink node in to
That's exactly what my question was: since there is an external producer in
the application without a sink node in topology, how will the streams know
that task is completed before committing the offset or it will not know at
all ?
Second question is: can there be multiple sink nodes i
> Matthias,
>>>
>>> Do you have any inputs on above queries?
>>>
>>> On Wed, Jun 30, 2021 at 7:15 PM Pushkar Deole
>> wrote:
>>>
>>>> Hi,
>>>>
>>>> Our application uses kafka streams that reads from a source t
t; Matthias,
> >
> > Do you have any inputs on above queries?
> >
> > On Wed, Jun 30, 2021 at 7:15 PM Pushkar Deole
> wrote:
> >
> >> Hi,
> >>
> >> Our application uses kafka streams that reads from a source topic, does
> >> pr
gt; Do you have any inputs on above queries?
>
> On Wed, Jun 30, 2021 at 7:15 PM Pushkar Deole wrote:
>
>> Hi,
>>
>> Our application uses kafka streams that reads from a source topic, does
>> processing on records and produces processed record on destination to
Matthias,
Do you have any inputs on above queries?
On Wed, Jun 30, 2021 at 7:15 PM Pushkar Deole wrote:
> Hi,
>
> Our application uses kafka streams that reads from a source topic, does
> processing on records and produces processed record on destination topic
> through the
Hi,
Our application uses kafka streams that reads from a source topic, does
processing on records and produces processed record on destination topic
through the use of external producer i.e. the producer created via kafka
producer API.
Does this model still guarantee exactly once semantic or it
o
On 2021/05/17 15:35:12, mangat rai wrote:
> Urko,>
>
> You can enable changelog topics for your state store. This will enable
the>
> application to persist the data to a Kafka topic. Next time when>
> application start, it will first build it's state by using this topic.
Ar
Urko,
You can enable changelog topics for your state store. This will enable the
application to persist the data to a Kafka topic. Next time when
application start, it will first build it's state by using this topic. Are
you using Kstreams or the low-level processor API?
Regards,
Mangat
O
Hello,
I have a question regarding the use of Kafka/Kafka Streams to store the
state of a stateful application.
My application is filtering on a stream based on a value from the previous
event of the stream. For example, if the previous car with the same model
was red, this car cannot be red
t; Hello,
>
>
>
> I’m working on creating an application that leverages Kafka and Kafka
> Streams. I have some issues with application startup that I’ve been unable
> to solve myself even with the help of my team mates, so I’m writing here in
> the hopes that someone could off
Hello,
I’m working on creating an application that leverages Kafka and Kafka Streams.
I have some issues with application startup that I’ve been unable to solve
myself even with the help of my team mates, so I’m writing here in the hopes
that someone could offer help. I’d be very grateful
elog topic.
Best,
Sophie
On Thu, Dec 17, 2020 at 11:59 AM Ming Liu wrote:
> Hi Team,
> I can't find any documentation or guidance on the expectation of the
> downgrade of stateful Kafka stream application (which has different rocksdb
> versions embedded).
> For example,
Hi Team,
I can't find any documentation or guidance on the expectation of the
downgrade of stateful Kafka stream application (which has different rocksdb
versions embedded).
For example, if we upgrade from 2.2 to 2.5 (with binary upgrade only
and using the same eager protocol) and so
Don't worry about this one. Figured it out. Created a Serde for MyAppObject
& implemented my own Serializer & Deserializer classes.
object MyAppObjectSerde extends Serde[MyAppObject] {
override def serializer(): Serializer[MyAppObject] = new MyAppObjectSerializer
override def deserializer():
I am grouping messages as follows:
.groupBy((_, myAppObject) => myAppObject.getId)(Grouped.`with`[String,
MyAppObject])
I am getting a message saying.. "No implicits found for parameter
valueSerde: Serde[MyAppObject]
My understanding is I need to add an implicit like this...
implicit val *my
Scheidegger
Sent: Sunday, November 15, 2020 9:57 AM
To: users@kafka.apache.org
Subject: Re: On the best number of partitions per topic for kafka based
application and the best number of consumers per topic?
Kafka alone wouldn't scale down your consumers, but you can get metrics
from Kafka and use
wrote:
> Thanks,
> My question is mostly about dynamic resource optimization,
>
> say I configured my application with 30 partitions and then I managed for
> 30 consumers (within a consumer group) to read/process the produced
> messages, but say at for instance at some unpeak
Thanks,
My question is mostly about dynamic resource optimization,
say I configured my application with 30 partitions and then I managed for 30
consumers (within a consumer group) to read/process the produced messages, but
say at for instance at some unpeak load, I realized that a single
This depends on the design of your application and how you are using Kafka.
For instance, if you are using Kafka as a message queuing app, using
consumers within a consumer group to load balance, then you should create
the topics with as many partitions as the max number of consumers within
the
Given a business application that resorts into a message queue solution like
Kafka, what is the best number of partitions to select for a given topic? what
influences such a decision?
On the other hand, say we want to achieve a maximal throughput of message
consumption but at minimal resource
ues.apache.org/jira/browse/KAFKA-10689
>
> I also think we can do a better job of surfacing issues like this,
> rather than letting the
> application silently spin without making progress. I left some thoughts on
> the JIRA
> ticket and will try to incorporate one of them into th
Ok I looked into this a bit and found the bug. I'll open a PR with the fix
sometime today:
https://issues.apache.org/jira/browse/KAFKA-10689
I also think we can do a better job of surfacing issues like this,
rather than letting the
application silently spin without making progress. I left
rt of the task assignment process:
> >
> > 2020-10-28 12:22:37.879 DEBUG 27226 --- [-StreamThread-1]
> > o.a.k.s.p.i.StreamsPartitionAssignor : stream-thread
> >
> >
> [demo-application-81060bdc-c8cc-4350-85f8-d238267e264e-StreamThread-1-consumer]
> > Constr
> 2020-10-28 12:22:37.879 DEBUG 27226 --- [-StreamThread-1]
> o.a.k.s.p.i.StreamsPartitionAssignor : stream-thread
>
> [demo-application-81060bdc-c8cc-4350-85f8-d238267e264e-StreamThread-1-consumer]
> Constructed client metadata
> {81060bdc-c8cc-4350-85f8-d238267e264e=ClientMetadata{hostInf
ead-1]
o.a.k.s.p.i.StreamsPartitionAssignor : stream-thread
[demo-application-81060bdc-c8cc-4350-85f8-d238267e264e-StreamThread-1-consumer]
Constructed client metadata
{81060bdc-c8cc-4350-85f8-d238267e264e=ClientMetadata{hostInfo=null,
consumers=[demo-application-81060bdc-c8cc-4350-85f8-d238267e264e-StreamThr
at's probably
the reason. It's not at all uncommon for restoration to take more than 30
seconds.
If it really is rebalancing this entire time, then you need to look into
the logs to figure
out why. I don't see anything obviously wrong with your particular
application, and even
if ther
Hey there!
My team and I have run across a bit of a jam in our application where,
given a particular setup, our Kafka Streams application never seems to
start successfully, instead just getting stuck in the REBALANCING state.
We've been able to get the crucial factors that cause this beh
links Gilles posted.
>
> Best,
> Bruno
>
> On 22.09.20 10:51, Pushkar Deole wrote:
> > Thank you Gilles..will take a look..
> >
> > Bruno, thanks for your elaborate explanation as well... however it
> > basically exposes my application to certain issues..
>
it
basically exposes my application to certain issues..
e.g. the application deals with agent states of a call center, and where
the order of processing is important. So when agent is logged in then he
keeps rotating between Ready, and Not ready states and at the end of the
day he becomes Logged
Thank you Gilles..will take a look..
Bruno, thanks for your elaborate explanation as well... however it
basically exposes my application to certain issues..
e.g. the application deals with agent states of a call center, and where
the order of processing is important. So when agent is logged in
te:
> > Bruno,
> >
> > So, essentially, we are just waiting on the processing of first event
> that
> > got an error before going ahead on to the next one.
> >
> > Second, if application handles storing the events in state store for
> retry,
> > Ka
application handles storing the events in state store for retry,
Kafka stream would essentially commit the offset of those events, so next
event will be polled by consumer, correct?
Instead of this work around, is there any provision in kafka streams for
this scenario? e.g. in case application
Bruno,
So, essentially, we are just waiting on the processing of first event that
got an error before going ahead on to the next one.
Second, if application handles storing the events in state store for retry,
Kafka stream would essentially commit the offset of those events, so next
event will
21.09.20 14:21, Pushkar Deole wrote:
> > Bruno,
> >
> > 1. the loading of topic mapped to GlobalKTable is done by some other
> > service/application so when my application starts up, it will just sync a
> > GlobalKTable against that topic and if that other service/appl
stopped and nothing is emitted.
Best,
Bruno
On 21.09.20 14:21, Pushkar Deole wrote:
Bruno,
1. the loading of topic mapped to GlobalKTable is done by some other
service/application so when my application starts up, it will just sync a
GlobalKTable against that topic and if that other service
Bruno,
1. the loading of topic mapped to GlobalKTable is done by some other
service/application so when my application starts up, it will just sync a
GlobalKTable against that topic and if that other service/application is
still starting up then it may not have loaded that data on that topic and
once. You also need to avoid the state store growing
indefinitely if required data in the global table is not available for a
long time or not available at all. Maybe all this caveats do not apply
to your use case.
Best,
Bruno
On 21.09.20 13:45, Pushkar Deole wrote:
Say the application level
Say the application level exception is named as :
MeasureDefinitionNotAvaialbleException
What I am trying to achieve is: in above case when the event processing
fails due to required data not available, the streams should not proceed on
to next event, however it should wait for the processing of
It is not a kafka streams error, it is an application level error e.g. say,
some data required for processing an input event is not available in the
GlobalKTable since it is not yet synced with the global topic
On Mon, Sep 21, 2020 at 4:54 PM Bruno Cadonna wrote:
> Hi Pushkar,
>
> Is
Hi Pushkar,
Is the error you are talking about, one that is thrown by Kafka Streams
or by your application? If it is thrown by Kafka Streams, could you
please post the error?
I do not completely understand what you are trying to achieve, but maybe
max.task.idle.ms [1] is the configuration
Hi,
I would like to know how to handle following scenarios while processing
events in a kafka streams application:
1. the streams application needs data from a globalKtable which loads it
from a topic that is populated by some other service/application. So, if
the streams application starts
Hi Matthias,
I have configured the GlobalKTable to stream from a topic and application
is working fine, however during automated build test cases, sometimes I get
an exception: I believe this could be because of race between actual topic
creation and the service startup (since topic creation may
Hello,
I'm playing around kafka for building a chat application. Here is what I
have done so far:
1. I have set up a CDC (change data capture) on my Postgresql database
2. The change on my table will get published to Kafka
3. I have node socket.io server, which listen to these messages on
It's self-service.
Please follow the instructions from the web page:
https://kafka.apache.org/contact
-Matthias
On 6/7/20 4:49 PM, wang120445...@sina.com wrote:
> hi kafka:
>I want to apply Contributor's List.
> end
>
>
>
> wang120445...@sina.com
>
hi kafka:
I want to apply Contributor's List.
end
wang120445...@sina.com
c up with topic
>> after
>> >>> the event has been put on the topic.
>> >>>
>> >>> On Tue, May 26, 2020 at 10:58 PM Matthias J. Sax
>> >>> wrote:
>> >>>
>> >>>> For example it could be some "sta
nformation, like a mapping from
> >>>> zip code to city name.
> >>>>
> >>>> Something that does usually not change over time.
> >>>>
> >>>>
> >>>> -Matthias
> >>>>
> >>>> On 5/25/20
t;>>>
>>>>
>>>> -Matthias
>>>>
>>>> On 5/25/20 9:55 PM, Pushkar Deole wrote:
>>>>> Matthias,
>>>>>
>>>>> I am wondering what you mean by "Global store hold "axially"
as,
>>> >
>>> > I am wondering what you mean by "Global store hold "axially" data that
>>> is
>>> > provided from "outside" of the
>>> > app"
>>> >
>>> > will you be able to give some example u
able to give some example use case here as to what you mean
>> by
>> > axially data provided from outside app?
>> >
>> > On Sat, May 2, 2020 at 1:58 AM Matthias J. Sax
>> wrote:
>> >
>> >> Both stores sever a different purpose.
>> >&g
t; > app"
> >
> > will you be able to give some example use case here as to what you mean
> by
> > axially data provided from outside app?
> >
> > On Sat, May 2, 2020 at 1:58 AM Matthias J. Sax wrote:
> >
> >> Both stores sever a different pu
Both stores sever a different purpose.
>>
>> Regular stores allow you to store state the application computes.
>> Writing into the changelog is a fault-tolerance mechanism.
>>
>> Global store hold "axially" data that is provided from "outside" of the
1:58 AM Matthias J. Sax wrote:
> Both stores sever a different purpose.
>
> Regular stores allow you to store state the application computes.
> Writing into the changelog is a fault-tolerance mechanism.
>
> Global store hold "axially" data that is provided from "outside
> >
> > I hope this helps!
> > -John
> >
> > On Tue, May 5, 2020, at 01:23, Pushkar Deole wrote:
> > > Thanks John... appreciate your inputs and suggestions. I have been
> > assigned
> > > recently to this task (of persisting the cache)
y to this task (of persisting the cache) and haven't been involved
> > in original design and architecture and agree with all the issues you
> have
> > highlighted.
> > However, at this point, i don't think the application can be converted to
> > streams since the d
Hello John, Matthias
Sorry for bothering you, however this is now getting crazier. Initially I
was under the impression that the cache being hold by application is in the
form of key/value where key is the instance of agentId (e.g. 10) and value
will hold other attributes (and their respective
:
> Thanks John... appreciate your inputs and suggestions. I have been assigned
> recently to this task (of persisting the cache) and haven't been involved
> in original design and architecture and agree with all the issues you have
> highlighted.
> However, at this point, i don't
Thanks John... appreciate your inputs and suggestions. I have been assigned
recently to this task (of persisting the cache) and haven't been involved
in original design and architecture and agree with all the issues you have
highlighted.
However, at this point, i don't think the applicat
is not designed to provide replicated
> > caching mechanism wherein the updates to cache will be synchronous across
> > multiple cache instances.
> >
> > On Sun, May 3, 2020 at 10:49 PM Pushkar Deole
> wrote:
> >
> > > Thanks John.
> > >
> > >
instances.
>
> On Sun, May 3, 2020 at 10:49 PM Pushkar Deole wrote:
>
> > Thanks John.
> >
> > Actually, this is a normal consumer-producer application wherein there are
> > 2 consumers (admin consumer and main consumer) consuming messages from 2
> > dif
-producer application wherein there are
> 2 consumers (admin consumer and main consumer) consuming messages from 2
> different topics.
> One of the consumers consumes messages from a admin topic and populates
> data in a cache e.g. lets say agent with agent id 10 for which the first
> name a
Thanks John.
Actually, this is a normal consumer-producer application wherein there are
2 consumers (admin consumer and main consumer) consuming messages from 2
different topics.
One of the consumers consumes messages from a admin topic and populates
data in a cache e.g. lets say agent with agent
Hi Pushkar,
I’ve been wondering if we should add writable tables to the Streams api. Can
you explain more about your use case and how it would integrate with your
application?
Incidentally, this would also help us provide more concrete advice.
Thanks!
John
On Fri, May 1, 2020, at 15:28
Both stores sever a different purpose.
Regular stores allow you to store state the application computes.
Writing into the changelog is a fault-tolerance mechanism.
Global store hold "axially" data that is provided from "outside" of the
app. There is no changelog topic, but
> >>>
> >>> I am wondering if this is possible: i have been asked to use state
> stores
> >>> as a general replicated cache among multiple instances of service
> >> instances
> >>> however the state store is created through streambuilder but i
e is created through streambuilder but is not
>>> actually modified through stream processor topology however it is to be
>>> modified from outside the stream topology. So, essentially, the state
>> store
>>> is just to be created from streambuilder and then to
lder but is not
> > actually modified through stream processor topology however it is to be
> > modified from outside the stream topology. So, essentially, the state
> store
> > is just to be created from streambuilder and then to be used as an
> > application level cach
ugh stream processor topology however it is to be
> modified from outside the stream topology. So, essentially, the state store
> is just to be created from streambuilder and then to be used as an
> application level cache that will get replicated between application
> instances.
modified from outside the stream topology. So, essentially, the state store
is just to be created from streambuilder and then to be used as an
application level cache that will get replicated between application
instances. Is this possible using state stores?
Secondly, if possible, is this a good
Hi All,
I have a kafka application which consums events from multiple topics,
stores certain fields from the events in local in-memory cache, based on
the cached data it enriches some of the events consumed from another topic
e.g. say for agent id 10, the first name and last name are cached then
Hello,
As far as I know Kafka connect's Rest server depends on Glassfish for
web-app configuration; the Kafka brokers do not.
Guozhang
On Sat, Feb 22, 2020 at 5:29 AM ashish sood wrote:
> Hello,
>
> This is more of a query rather than an issue. I am working with java based
Hello,
This is more of a query rather than an issue. I am working with java based
application & Kafka connect worker in distributed mode. While going through
the logs post I started the worker, I see some logs for Glassfish. I would
want to know if Kafka uses Glassfish server internally to
You need to partition the input data correctly, thus that all records
with the same key go the same partition. For this case, all records with
the same key will be processed by the same task, and thus each key is
stored in one shard only.
-Matthias
On 11/15/19 4:28 PM, Gioacchino Vino wrote:
> Hi
Hi expert,
I don't understand a kafka behavior and I'm here to ask for explanation.
My processing task is pretty simple and it's quite similar to a
change-log one.
The record value contains a key/value pair: if the new value is
different respect the stored one, forward to the output topic a
> Why that? Just because there is explicit documentation?
Just that they target YARN.
Ryanne
On Thu, Nov 14, 2019, 1:59 AM Matthias J. Sax wrote:
> Why that? Just because there is explicit documentation?
>
>
> @Debraj: Kafka Streams can be deployed as a regular Java applicatio
Why that? Just because there is explicit documentation?
@Debraj: Kafka Streams can be deployed as a regular Java application.
Hence, and tutorial on how to run a Java application on YARN should help.
-Matthias
On 11/11/19 10:33 AM, Ryanne Dolan wrote:
> Consider using Flink, Spark, or Sa
Consider using Flink, Spark, or Samza instead.
Ryanne
On Fri, Nov 8, 2019, 4:27 AM Debraj Manna wrote:
> Hi
>
> Is there any documentation or link I can refer to for the steps for
> deploying the Kafka Streams application in YARN?
>
> Kafka Client - 0.11.0.3
> Kafka
1 - 100 of 354 matches
Mail list logo
