Kafka

Hi, I have a KStreams app that outputs a KTable to a topic with cleanup policy "compact,delete". I have the Confluent S3 Connector to store this table in S3 where I do further analysis with hive. Now my question is, if there's a way to trigger log compaction right before the S3 Connector reads the data so I store less data in S3 then when it simply copies all data from the stream? Thanks, Daniel

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 then he keeps rotating between Ready, and Not ready states and at the end of the day he becomes Logged out... If while processing the Ready event, there is some temporary issue with database/network and the event processing gets exception, application does few retries but no luck. As per kafka polling, it will go ahead and poll next record from partition for the same agent (since agent id being key) and it will process logged out event. So, this mean i lost the Ready event in between due to the database issue? Even if i store this event somewhere for processing it later, processing the Ready event after logged out, doesn't make sense since order of state is important? Is my ...

Hi Pushkar, Uber has written about how they deal with failures and reprocessing here, it might help you achieve what you describe: https://eng.uber.com/reliable-reprocessing/ . Unfortunately, there isn't much written documentation about those patterns. There's also a good talk from Confluent's Antony Stubbs on how you can do certain things with the Processor API that you can't do with the Kafka Streams DSL: https://www.confluent.io/kafka-summit-lon19/beyond-dsl-unlocking-power-kafka-streams-processor-api . Gilles Philippart Funding Circle Engineering On Tue, 22 Sep 2020 at 08:12, Bruno Cadonna < bruno@confluent.io > wrote: > Hi Pushkar, > > I think there is a misunderstanding. If a consumer polls from a > partition, it will always poll the next event independently whether the > offset was committed or not. Committed offsets are used for fault > tolerance, i.e., when a consumer crashes, the consumer that takes over ...

Yes, I use connect-mirror-maker.sh. On 2020/09/21 22:12:13, Ryanne Dolan < ryannedolan@gmail.com > wrote: > Oleg, yes you can run multiple MM2s for multiple DCs, and generally that's > what you want to do. Are you using Connect to run MM2, or the > connect-mirror-maker.sh driver? > > Ryanne > > On Mon, Sep 21, 2020, 3:38 PM Oleg Osipov < oleg.alex.osipov@gmail.com > > wrote: > > > I use the configuration for M2M for both datacentres > > clusters: > > - {"name": "dc1", "bootstrapServers": ip1} > > - {"name": "dc2", "bootstrapServers": ip2} > > > > Do you mean I need use additional names besides 'dc1' and 'dc2'? > > > > On 2020/09/21 17:27:50, nitin agarwal < nitingarg456@gmail.com > wrote: > > > Did you keep the cluster name the same ? If yes, then it will cause > ...

Hi Pushkar, I think there is a misunderstanding. If a consumer polls from a partition, it will always poll the next event independently whether the offset was committed or not. Committed offsets are used for fault tolerance, i.e., when a consumer crashes, the consumer that takes over the work of the crashed consumer will start polling record from the offset the crashed consumer committed last. This is not only true for Kafka Streams, but for all applications that use a Kafka consumer with subscription. To be clear, my proposal is not a workaround. This is one approach to solve your problem in Kafka Streams. You could have a look into stream-stream joins if you can use a stream instead of a global table. Another approach would be to use a plain Kafka consumer instead of Kafka Stream with which you would have a more fine-grained control about polls and commits. In any case, be aware that blocking processing on an event indefinitely may result in your lag a...

Hi everyone, I am using camel KAFKA inside Apache KARAF and producing messages to the KAFKA server. I have multiple interfaces configured in my system and sending(producing) messages will happen only by using the first interface *managmentserver1_local_interface* always. Same thing i tried with camel-ftp in which i can choose the network interface by using the name "*managmentserver1_traffic_interface*" and binding through the *bindAddress* api which is introduced after 2.23. Is there any way to configure the camel kafka component to select network interfaces? Currently my machine has 2 network interfaces. 127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4192.198.10.1 testserver1 managementserver1_local_interface192.199.11.2 testserver1 managementserver1_traffic_interface I Went through the source code of kafka and found that internally it is creating a socket which is used for transferring which will provide the o...

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 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 registers application level exceptions then kafka streams will take care of it and do all this internally, and will not commit the offset of that event and hence will keep polling the same event again? Since this is a common scenario, using a particular configuration for users can achieve this in Kafka streams internally? On Mon, Sep 21, 2020 at 9:01 PM Bruno Cadonna < bruno@confluent.io > wrote: > Hi Pushkar, > > If you want to keep the order, you could still use the state store I > suggested in my previous...