You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "Xiaolin Jia (JIRA)" <ji...@apache.org> on 2019/04/25 07:53:00 UTC
[jira] [Created] (KAFKA-8289) KTable, Long> can't
be suppressed
Xiaolin Jia created KAFKA-8289:
----------------------------------
Summary: KTable<Windowed<String>, Long> can't be suppressed
Key: KAFKA-8289
URL: https://issues.apache.org/jira/browse/KAFKA-8289
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 2.2.0
Environment: Broker on a Linux, stream app on my win10 laptop.
I add one row log.message.timestamp.type=LogAppendTime to my broker's server.properties. stream app all default config.
Reporter: Xiaolin Jia
I encountered a problem yesterday that I got more than one [Window Final Results|https://kafka.apache.org/22/documentation/streams/developer-guide/dsl-api.html#id31] from a session time window.
time ticker A -> (4,A) / 25s,
time ticker B -> (4, B) / 25s all send to the same topic
below is my stream app code
{code:java}
kstreams[0]
.peek((k, v) -> log.info("--> ping, k={},v={}", k, v))
.groupBy((k, v) -> v, Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(SessionWindows.with(Duration.ofSeconds(100)).grace(Duration.ofMillis(20)))
.count()
.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
.toStream().peek((k, v) -> log.info("window={},k={},v={}", k.window(), k.key(), v));
{code}
{{here is my log print}}
{noformat}
2019-04-24 20:00:26.142 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:00:47.070 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556106587744, endMs=1556107129191},k=A,v=20
2019-04-24 20:00:51.071 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:01:16.065 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:01:41.066 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:02:06.069 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:02:31.066 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:02:56.208 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:03:21.070 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:03:46.078 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:04:04.684 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:04:11.069 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:04:19.371 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107426409},k=B,v=9
2019-04-24 20:04:19.372 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107445012},k=A,v=1
2019-04-24 20:04:29.604 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:04:36.067 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:04:49.715 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107451397},k=B,v=10
2019-04-24 20:04:49.716 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107469935},k=A,v=2
2019-04-24 20:04:54.593 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:05:01.070 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:05:19.599 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:05:20.045 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107476398},k=B,v=11
2019-04-24 20:05:20.047 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107501398},k=B,v=12
2019-04-24 20:05:26.075 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:05:44.598 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:05:50.399 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107519930},k=A,v=4
2019-04-24 20:05:50.400 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107526405},k=B,v=13
2019-04-24 20:05:51.067 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:06:09.595 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:06:16.089 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:06:20.765 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107544929},k=A,v=5
2019-04-24 20:06:20.767 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107569926},k=A,v=6
2019-04-24 20:06:34.595 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:06:41.063 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:06:51.081 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107576415},k=B,v=15
2019-04-24 20:06:51.082 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107594925},k=A,v=7
2019-04-24 20:06:59.607 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:07:06.072 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:07:21.440 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107601391},k=B,v=16
2019-04-24 20:07:21.441 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107619935},k=A,v=8
2019-04-24 20:07:24.596 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:07:31.066 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:07:49.608 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:07:51.775 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107626420},k=B,v=17
2019-04-24 20:07:51.777 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107651396},k=B,v=18
2019-04-24 20:07:56.064 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:08:14.591 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:08:21.066 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:08:22.125 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107669943},k=A,v=10
2019-04-24 20:08:22.126 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107694921},k=A,v=11
2019-04-24 20:08:39.619 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:08:46.067 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:08:52.457 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107701397},k=B,v=20
2019-04-24 20:08:52.458 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107719949},k=A,v=12
2019-04-24 20:09:04.599 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=A
2019-04-24 20:09:11.066 INFO --- [-StreamThread-1] c.g.k.AppStreams : --> ping, k=4,v=B
2019-04-24 20:09:22.794 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107226473, endMs=1556107726398},k=B,v=21
2019-04-24 20:09:22.796 INFO --- [-StreamThread-1] c.g.k.AppStreams : window=Window{startMs=1556107445012, endMs=1556107744928},k=A,v=13{noformat}
My source stream record rate is a same v record / 25s, gap of inactivity is 100s, 25<100. I shouldn't got so many window info print log. Session window close and then reopened? my grace is 20 millisecond, it should not be ah.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)