Hi Mangat, thank you for clarification. I'll try the suggested configuration as soon as I get back to work. Will keep you posted then.
Best wishes Karsten mangat rai <mangatm...@gmail.com> schrieb am Mi., 27. März 2024, 11:07: > Hey Karsten, > > You don't need to do any other configuration to enable EOS. See here - > > https://docs.confluent.io/platform/current/streams/concepts.html#processing-guarantees > It mentions that the producer will be idempotent. That also mans ack=all > will be considered. Not that if you have any other ack from the config, it > will be ignored in the favour of exactly-once. > > Do let me know if that solves your problem. I am curious. if yes, then I > would ask you to create an issue. > > Regards, > Mangat > > On Wed, Mar 27, 2024 at 10:49 AM Karsten Stöckmann < > karsten.stoeckm...@gmail.com> wrote: > > > Hi Mangat, > > > > thanks for clarification. So to my knowledge exactly-once is configured > > using the 'processing.guarantee=exactly_once_v2' setting? Is the > > configuration setting 'acks=all' somehow related and would you advise > > setting that as well? > > > > Best wishes > > Karsten > > > > > > mangat rai <mangatm...@gmail.com> schrieb am Di., 26. März 2024, 15:44: > > > > > Hey Karsten, > > > > > > So if a topic has not been created yet. Streams app will keep the data > in > > > memory, and then write it later when it is available. if your app is > > > restarted (or thread is killed), you may lose data but it depends if > the > > > app will commit in the source topics. If there is no errors, then it > > should > > > be persisted eventually. > > > > > > However, overall exactly-once provides a much tighter and better commit > > > control. If you don't have scaling issue, I will strongly advise you to > > use > > > EOS. > > > > > > Thanks, > > > Mangat > > > > > > > > > On Tue, Mar 26, 2024 at 3:33 PM Karsten Stöckmann < > > > karsten.stoeckm...@gmail.com> wrote: > > > > > > > Hi Mangat, > > > > > > > > thanks for your thoughts. I had actually considered exactly-once > > > semantics > > > > already, was unsure whether it would help, and left it aside for once > > > then. > > > > I'll try that immediately when I get back to work. > > > > > > > > About snapshots and deserialization - I doubt that the issue is > caused > > by > > > > deserialization failures because: when taking another (i.e. at a > later > > > > point of time) snapshot of the exact same data, all messages fed into > > the > > > > input topic pass the pipeline as expected. > > > > > > > > Logs of both Kafka and Kafka Streams show no signs of notable issues > as > > > far > > > > as I can tell, apart from these (when initially starting up, > > intermediate > > > > topics not existing yet): > > > > > > > > 2024-03-22 22:36:11,386 WARN [org.apa.kaf.cli.NetworkClient] > > > > > > > > > > > > > > (kstreams-folder-aggregator-a38397c2-d30a-437e-9817-baa605d49e23-StreamThread-4) > > > > [Consumer > > > > > > > > > > > > > > clientId=kstreams-folder-aggregator-a38397c2-d30a-437e-9817-baa605d49e23-StreamThread-4-consumer, > > > > groupId=kstreams-folder-aggregator] Error while fetching metadata > with > > > > correlation id 69 : > > > > > > > > > > > > > > {kstreams-folder-aggregator-folder-to-agency-subscription-response-topic=UNKNOWN_TOPIC_OR_PARTITION, > > > > <various other intermediate topics>} > > > > > > > > Best wishes > > > > Karsten > > > > > > > > > > > > > > > > mangat rai <mangatm...@gmail.com> schrieb am Di., 26. März 2024, > > 11:06: > > > > > > > > > Hey Karsten, > > > > > > > > > > There could be several reasons this could happen. > > > > > 1. Did you check the error logs? There are several reasons why the > > > Kafka > > > > > stream app may drop incoming messages. Use exactly-once semantics > to > > > > limit > > > > > such cases. > > > > > 2. Are you sure there was no error when deserializing the records > > from > > > > > `folderTopicName`. You mentioned that it happens only when you > start > > > > > processing and the other table snapshot works fine. This gives me a > > > > feeling > > > > > that the first records in the topic might not be deserialized > > properly. > > > > > > > > > > Regards, > > > > > Mangat > > > > > > > > > > On Tue, Mar 26, 2024 at 8:45 AM Karsten Stöckmann < > > > > > karsten.stoeckm...@gmail.com> wrote: > > > > > > > > > > > Hi, > > > > > > > > > > > > thanks for getting back. I'll try and illustrate the issue. > > > > > > > > > > > > I've got an input topic 'folderTopicName' fed by a database CDC > > > system. > > > > > > Messages then pass a series of FK left joins and are eventually > > sent > > > to > > > > > an > > > > > > output topic like this ('agencies' and 'documents' being > KTables): > > > > > > > > > > > > > > > > > > streamsBuilder // > > > > > > .table( // > > > > > > folderTopicName, // > > > > > > Consumed.with( // > > > > > > folderKeySerde, // > > > > > > folderSerde)) // > > > > > > .leftJoin( // > > > > > > agencies, // > > > > > > Folder::agencyIdValue, // > > > > > > AggregateFolder::new, // > > > > > > TableJoined.as("folder-to-agency"), > // > > > > > > Materializer // > > > > > > .<FolderId, > > > > > > AggregateFolder>named("folder-to-agency-materialized") // > > > > > > .withKeySerde(folderKeySerde) > > // > > > > > > > > > > > .withValueSerde(aggregateFolderSerde)) > > > > > > // > > > > > > .leftJoin( // > > > > > > documents, // > > > > > > .toStream(... > > > > > > .to(... > > > > > > > > > > > > ... > > > > > > > > > > > > As far as I understand, left join sematics should be similar to > > those > > > > of > > > > > > relational databases, i.e. the left hand value always passes the > > join > > > > > with > > > > > > the right hand value set as <null> if not present. Whereas what I > > am > > > > > > observing is this: lots of messages on the input topic are even > > > absent > > > > on > > > > > > the first left join changelog topic > > > > > > ('folder-to-agency-materialized-changelog'). But: this seems to > > > happen > > > > > only > > > > > > in case the Streams application is fired up for the first time, > > i.e. > > > > > > intermediate topics do not yet exist. When streaming another > table > > > > > snapshot > > > > > > to the input topic, things seem (!) to work as expected... > > > > > > > > > > > > Best wishes, > > > > > > Karsten > > > > > > > > > > > > Bruno Cadonna <cado...@apache.org> schrieb am Mo., 25. März > 2024, > > > > 17:01: > > > > > > > > > > > > > Hi, > > > > > > > > > > > > > > That sounds worrisome! > > > > > > > > > > > > > > Could you please provide us with a minimal example that shows > the > > > > issue > > > > > > > you describe? > > > > > > > > > > > > > > That would help a lot! > > > > > > > > > > > > > > Best, > > > > > > > Bruno > > > > > > > > > > > > > > On 3/25/24 4:07 PM, Karsten Stöckmann wrote: > > > > > > > > Hi, > > > > > > > > > > > > > > > > are there circumstances that might lead to messages silently > > > (i.e. > > > > > > > without > > > > > > > > any logged warnings or errors) disappearing from a topology? > > > > > > > > > > > > > > > > Specifically, I've got a rather simple topology doing a > series > > of > > > > FK > > > > > > left > > > > > > > > joins and notice severe message loss in case the application > is > > > > fired > > > > > > up > > > > > > > > for the first time, i.e. intermediate topics not existing > yet. > > > I'd > > > > > > > > generally expect the message count on the output topic > resemble > > > > that > > > > > > from > > > > > > > > the input topic, yet it doesn't (about half only). > > > > > > > > > > > > > > > > Any hints on this? > > > > > > > > > > > > > > > > Best wishes > > > > > > > > Karsten > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
