Hi Jonathan, Just a quick update: I have not been able to reproduce the duplicates issue with the 2.2 RC, even with a topology very similar to the one you included in your stackoverflow post.
I think we should treat this as a new bug. Would you mind opening a new Jira bug ticket with some steps to reproduce the problem, and also exactly the behavior you observe? Thanks, -John On Mon, Mar 4, 2019 at 10:41 PM John Roesler <j...@confluent.io> wrote: > Hi Jonathan, > > Sorry to hear that the feature is causing you trouble as well, and that > the 2.2 release candidate didn't seem to fix it. > > I'll try and do a repro based on the code in your SO post tomorrow. > > I don't think it's related to the duplicates, but that shutdown error is > puzzling. Can you print the topology (with topology.describe() ) ? This > will tell us what is in task 1 (i.e., *1_*) of your program. > > Thanks, > -John > > On Fri, Mar 1, 2019 at 11:33 AM Jonathan Santilli < > jonathansanti...@gmail.com> wrote: > >> BTW, after stopping the app gracefully (Stream#close()), this error shows >> up repeatedly: >> >> 2019-03-01 17:18:07,819 WARN >> [XXX-116ba7c8-678e-47f7-9074-7d03627b1e1a-StreamThread-1] >> internals.ProcessorStateManager (ProcessorStateManager.java:327) - task >> [0_0] Failed to write offset checkpoint file to >> [/tmp/kafka-stream/XXX/0_0/.checkpoint] >> >> java.io.FileNotFoundException: /tmp/kafka-stream/XXX/0_0/.checkpoint.tmp >> (No such file or directory) >> >> at java.io.FileOutputStream.open0(Native Method) ~[?:1.8.0_191] >> >> at java.io.FileOutputStream.open(FileOutputStream.java:270) ~[?:1.8.0_191] >> >> at java.io.FileOutputStream.<init>(FileOutputStream.java:213) >> ~[?:1.8.0_191] >> >> at java.io.FileOutputStream.<init>(FileOutputStream.java:162) >> ~[?:1.8.0_191] >> >> at org.apache.kafka.streams.state.internals.OffsetCheckpoint.write( >> OffsetCheckpoint.java:79) ~[kafka-streams-2.2.0.jar:?] >> >> at >> >> org.apache.kafka.streams.processor.internals.ProcessorStateManager.checkpoint( >> ProcessorStateManager.java:325) [kafka-streams-2.2.0.jar:?] >> >> at org.apache.kafka.streams.processor.internals.StreamTask.suspend( >> StreamTask.java:599) [kafka-streams-2.2.0.jar:?] >> >> at org.apache.kafka.streams.processor.internals.StreamTask.close( >> StreamTask.java:721) [kafka-streams-2.2.0.jar:?] >> >> at org.apache.kafka.streams.processor.internals.AssignedTasks.close( >> AssignedTasks.java:337) [kafka-streams-2.2.0.jar:?] >> >> at org.apache.kafka.streams.processor.internals.TaskManager.shutdown( >> TaskManager.java:267) [kafka-streams-2.2.0.jar:?] >> >> at >> >> org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown( >> StreamThread.java:1209) [kafka-streams-2.2.0.jar:?] >> >> at org.apache.kafka.streams.processor.internals.StreamThread.run( >> StreamThread.java:786) [kafka-streams-2.2.0.jar:?] >> >> >> However, I have checked and the folder created starts with: *1_* >> >> ls -lha /tmp/kafka-stream/XXX/1_1 >> total 8 >> drwxr-xr-x 5 a b 160B 1 Mar 17:18 . >> drwxr-xr-x 34 a b 1.1K 1 Mar 17:15 .. >> -rw-r--r-- 1 a b 2.9K 1 Mar 17:18 .checkpoint >> -rw-r--r-- 1 a b 0B 1 Mar 16:05 .lock >> drwxr-xr-x 3 a b 96B 1 Mar 16:43 >> KSTREAM-REDUCE-STATE-STORE-0000000005 >> >> >> >> Cheers! >> -- >> Jonathan >> >> >> >> On Fri, Mar 1, 2019 at 5:11 PM Jonathan Santilli < >> jonathansanti...@gmail.com> >> wrote: >> >> > Hello John, hope you are well. >> > I have tested the version 2.2 release candidate (although I know it has >> > been postponed). >> > I have been following this email thread because I think am experiencing >> > the same issue. I have reported in an email to this list and also all >> the >> > details are in OS ( >> > >> https://stackoverflow.com/questions/54145281/why-do-the-offsets-of-the-consumer-group-app-id-of-my-kafka-streams-applicatio >> > ). >> > >> > After the test, the result is the same as before (at least for my case), >> > already processed records are passed again to the output topic causing >> the >> > data duplication: >> > >> > ... >> > 2019-03-01 16:55:23,808 INFO >> [XXX-116ba7c8-678e-47f7-9074-7d03627b1e1a-StreamThread-1] >> > internals.StoreChangelogReader (StoreChangelogReader.java:221) - >> > stream-thread [XXX-116ba7c8-678e-47f7-9074-7d03627b1e1a-StreamThread-1] >> No >> > checkpoint found for task 1_10 state store >> > KTABLE-SUPPRESS-STATE-STORE-0000000011 changelog >> > XXX-KTABLE-SUPPRESS-STATE-STORE-0000000011-changelog-10 with EOS turned >> on. *Reinitializing >> > the task and restore its state from the beginning.* >> > >> > ... >> > >> > >> > I was hoping for this to be fixed, but is not the case, at least for my >> > case. >> > >> > If you can, please take a look at the question in SO, I was in contact >> > with Matthias about it, he also points me the place where probably the >> > potential but could be present. >> > >> > Please, let me know any thoughts. >> > >> > >> > Cheers! >> > -- >> > Jonathan >> > >> > >> > On Tue, Feb 26, 2019 at 5:23 PM John Roesler <j...@confluent.io> wrote: >> > >> >> Hi again, Peter, >> >> >> >> Just to close the loop about the bug in Suppress, we did get the >> >> (apparent) >> >> same report from a few other people: >> >> https://issues.apache.org/jira/browse/KAFKA-7895 >> >> >> >> I also managed to reproduce the duplicate-result behavior, which could >> >> cause it to emit both intermediate results and duplicate final results. >> >> >> >> There's a patch for it in the 2.2 release candidate. Perhaps you can >> try >> >> it >> >> out and see if it resolves the issue for you? >> >> >> >> I'm backporting the fix to 2.1 as well, but I unfortunately missed the >> >> last >> >> 2.1 bugfix release. >> >> >> >> Thanks, >> >> -John >> >> >> >> On Fri, Jan 25, 2019 at 10:23 AM John Roesler <j...@confluent.io> >> wrote: >> >> >> >> > Hi Peter, >> >> > >> >> > Thanks for the replies. >> >> > >> >> > Regarding transactions: >> >> > Yes, actually, with EOS enabled, the changelog and the output topics >> are >> >> > all produced with the same transactional producer, within the same >> >> > transactions. So it should already be atomic. >> >> > >> >> > Regarding restore: >> >> > Streams doesn't put the store into service until the restore is >> >> completed, >> >> > so it should be guaranteed not to happen. But there's of course no >> >> > guarantee that I didn't mess something up. I'll take a hard look at >> it. >> >> > >> >> > Regarding restoration and offsets: >> >> > Your guess is correct: Streams tracks the latest stored offset >> outside >> >> of >> >> > the store implementation itself, specifically by writing a file >> (called >> >> a >> >> > Checkpoint File) in the state directory. If the file is there, it >> reads >> >> > that offset and restores from that point. If the file is missing, it >> >> > restores from the beginning of the stream. So it should "just work" >> for >> >> > you. Just for completeness, there have been several edge cases >> >> discovered >> >> > where this mechanism isn't completely safe, so in the case of EOS, I >> >> > believe we actually disregard that checkpoint file and the prior >> state >> >> and >> >> > always rebuild from the earliest offset in the changelog. >> >> > >> >> > Personally, I would like to see us provide the ability to store the >> >> > checkpoint inside the state store, so that checkpoint updates are >> >> > linearized correctly w.r.t. data updates, but I actually haven't >> >> mentioned >> >> > this thought to anyone until now ;) >> >> > >> >> > Finally, regarding your prior email: >> >> > Yes, I was thinking that the "wrong" output values might be part of >> >> > rolled-back transactions and therefore enabling read-committed mode >> on >> >> the >> >> > consumer might tell a different story that what you've seen to date. >> >> > >> >> > I'm honestly still baffled about those intermediate results that are >> >> > sneaking out. I wonder if it's something specific to your data >> stream, >> >> like >> >> > maybe if there is maybe an edge case when two records have exactly >> the >> >> same >> >> > timestamp? I'll have to stare at the code some more... >> >> > >> >> > Regardless, in order to reap the benefits of running the app with >> EOS, >> >> you >> >> > really have to also set your consumers to read_committed. Otherwise, >> >> you'll >> >> > be seeing output data from aborted (aka rolled-back) transactions, >> and >> >> you >> >> > miss the intended "exactly once" guarantee. >> >> > >> >> > Thanks, >> >> > -John >> >> > >> >> > On Fri, Jan 25, 2019 at 1:51 AM Peter Levart <peter.lev...@gmail.com >> > >> >> > wrote: >> >> > >> >> >> Hi John, >> >> >> >> >> >> Haven't been able to reinstate the demo yet, but I have been >> re-reading >> >> >> the following scenario of yours.... >> >> >> >> >> >> On 1/24/19 11:48 PM, Peter Levart wrote: >> >> >> > Hi John, >> >> >> > >> >> >> > On 1/24/19 3:18 PM, John Roesler wrote: >> >> >> > >> >> >> >> >> >> >> >> The reason is that, upon restart, the suppression buffer can only >> >> >> >> "remember" what got sent & committed to its changelog topic >> before. >> >> >> >> >> >> >> >> The scenario I have in mind is: >> >> >> >> >> >> >> >> ... >> >> >> >> * buffer state X >> >> >> >> ... >> >> >> >> * flush state X to buffer changelog >> >> >> >> ... >> >> >> >> * commit transaction T0; start new transaction T1 >> >> >> >> ... >> >> >> >> * emit final result X (in uncommitted transaction T1) >> >> >> >> ... >> >> >> >> * crash before flushing to the changelog the fact that state X >> was >> >> >> >> emitted. >> >> >> >> Also, transaction T1 gets aborted, since we crash before >> committing. >> >> >> >> ... >> >> >> >> * restart, restoring state X again from the changelog (because >> the >> >> emit >> >> >> >> didn't get committed) >> >> >> >> * start transaction T2 >> >> >> >> * emit final result X again (in uncommitted transaction T2) >> >> >> >> ... >> >> >> >> * commit transaction T2 >> >> >> >> ... >> >> >> >> >> >> >> >> So, the result gets emitted twice, but the first time is in an >> >> aborted >> >> >> >> transaction. This leads me to another clarifying question: >> >> >> >> >> >> >> >> Based on your first message, it seems like the duplicates you >> >> observe >> >> >> >> are >> >> >> >> in the output topic. When you read the topic, do you configure >> your >> >> >> >> consumer with "read committed" mode? If not, you'll see "results" >> >> from >> >> >> >> uncommitted transactions, which could explain the duplicates. >> >> >> >> >> >> ...and I was thinking that perhaps the right solution to the >> >> suppression >> >> >> problem would be to use transactional producers for the resulting >> >> output >> >> >> topic AND the store change-log. Is this possible? Does the >> compaction >> >> of >> >> >> the log on the brokers work for transactional producers as >> expected? In >> >> >> that case, the sending of final result and the marking of that fact >> in >> >> >> the store change log would together be an atomic operation. >> >> >> That said, I think there's another problem with suppression which >> looks >> >> >> like the supression processor is already processing the input while >> the >> >> >> state store has not been fully restored yet or something related... >> Is >> >> >> this guaranteed not to happen? >> >> >> >> >> >> And now something unrelated I wanted to ask... >> >> >> >> >> >> I'm trying to create my own custom state store. From the API I can >> see >> >> >> it is pretty straightforward. One thing that I don't quite >> understand >> >> is >> >> >> how Kafka Streams know whether to replay the whole change log after >> the >> >> >> store registers itself or just a part of it and which part (from >> which >> >> >> offset per partition). There doesn't seem to be any API point >> through >> >> >> which the store could communicate this information back to Kafka >> >> >> Streams. Is such bookkeeping performed outside the store? Does Kafka >> >> >> Streams first invoke flush() on the store and then notes down the >> >> >> offsets from the change log producer somewhere? So next time the >> store >> >> >> is brought up, the log is only replayed from last noted down >> offset? So >> >> >> it can happen that the store gets some log entries that have already >> >> >> been incorporated in it (from the point of one flush before) but >> never >> >> >> misses any... In any case there has to be an indication somewhere >> that >> >> >> the store didn't survive and has to be rebuilt from scratch. How do >> >> >> Kafka Streams detect that situation? By placing some marker file >> into >> >> >> the directory reserved for store's local storage? >> >> >> >> >> >> Regards, Peter >> >> >> >> >> >> >> >> >> > >> > >> > -- >> > Santilli Jonathan >> > >> >> >> -- >> Santilli Jonathan >> >
