I just tested out using kafka stream 2.3.0 and the issue went away.. Restarting the stream application did not cause all the messages to be reproduced again any more.. and the suppressor start producing from when it was shutdown.
Thanks for the help! Tao -----Original Message----- From: Tao Wang Sent: Tuesday, October 29, 2019 10:09 AM To: users@kafka.apache.org Subject: RE: kafka stream ktable with suppress operator Thanks Matthia for the reply. I am using kafka-stream-2.1.0 jar. I will try 2.3.1 and see if the issue still there. Tao -----Original Message----- From: Matthias J. Sax [mailto:matth...@confluent.io] Sent: Sunday, October 27, 2019 11:25 PM To: users@kafka.apache.org Subject: Re: kafka stream ktable with suppress operator What version are you using? We fixed couple of bugs in `suppress()` -- I would recommend to use latest 2.3.1 bug-fix release. -Matthia On 10/25/19 9:12 AM, Tao Wang wrote: > When using suppress operator with windowed Ktable, it looks like restarting > the kafka stream causes the aggregated messages from the SUPPRESS-STATE-STORE > published again.. > > Here is the sudo code .. anything I am missing or anything can be done to > avoid this .. > > > KTable<Windowed<String>, String> test = <KStreamObject> > .groupByKey() > .windowedBy(TimeWindows.of(Duration.ofSeconds(30)).grace(Duration.ofSeconds(3))) > .aggregate( > .... > , > > Materialized.<String,String,WindowStore<String,String>>as("aggregated-stream21-store") > .withRetention(Duration.ofMinutes(5)) > .with(Serdes.String(), Serdes.String()) > ) > .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())) > > .toStream() > > .to("<topic_out>") > > > > > > So when restarting the stream app, the <topic_out> will have duplicated > messages from a while back ... is this expected behavior ? > > Thanks, > Tao Wang > > > > > > ________________________________ > > This message may contain confidential information and is intended for > specific recipients unless explicitly noted otherwise. If you have reason to > believe you are not an intended recipient of this message, please delete it > and notify the sender. This message may not represent the opinion of > Intercontinental Exchange, Inc. (ICE), its subsidiaries or affiliates, and > does not constitute a contract or guarantee. Unencrypted electronic mail is > not secure and the recipient of this message is expected to provide > safeguards from viruses and pursue alternate means of communication where > privacy or a binding message is desired. > ________________________________ This message may contain confidential information and is intended for specific recipients unless explicitly noted otherwise. If you have reason to believe you are not an intended recipient of this message, please delete it and notify the sender. This message may not represent the opinion of Intercontinental Exchange, Inc. (ICE), its subsidiaries or affiliates, and does not constitute a contract or guarantee. Unencrypted electronic mail is not secure and the recipient of this message is expected to provide safeguards from viruses and pursue alternate means of communication where privacy or a binding message is desired.
