kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From John Roesler <j...@confluent.io>
Subject Re: KTable.suppress(Suppressed.untilWindowCloses) does not suppress some non-final results when the kafka streams process is restarted
Date Tue, 05 Mar 2019 04:41:29 GMT
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 <
jonathansantilli@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 <
> jonathansantilli@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 <john@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 <john@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.levart@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
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message