[
https://issues.apache.org/jira/browse/KAFKA-13290?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17413405#comment-17413405
]
Matthias J. Sax commented on KAFKA-13290:
-----------------------------------------
Your observation is behavior by-design. You are using `suppress` with
"until-window-close" strategy. A window closes only if stream-time advanced
beyond window-end plus grace-period. Note that stream-time is data driven, ie,
if no new data arrives, stream-time cannot advance (it's independent for
wall-clock time). Closing this as "not a problem". There is a ticket that
request a "wall-clock time" emit strategy for suppress() already
https://issues.apache.org/jira/browse/KAFKA-7748 in case you want to follow up
there. – Of course, feel free to follow up on this ticket if you have further
questions.
> My timeWindows last aggregated message never emit until a new message coming
> -----------------------------------------------------------------------------
>
> Key: KAFKA-13290
> URL: https://issues.apache.org/jira/browse/KAFKA-13290
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.6.2
> Environment: Development
> Reporter: Steve Zhou
> Priority: Major
>
> I have a Kafka stream event processing code which aggregates 1 minutes data.
> It works as expected if data comes continuously,
> If we stop producer, then i found the last aggregated message does not emit
> until new message coming.
>
> Following is my sample code, @Bean
> public KStream<String, AggregateMetricsFields> kStream(StreamsBuilder
> streamBuilder) {
> KStream<String, AggregateMetricsFields> aggregatedData = streamBuilder
> .stream(dataTopic, dataConsumed)
> .groupByKey(Grouped.with(
> stringSerde,
> aggregateValueSerde))
>
> .windowedBy(TimeWindows.of(windowDuration).grace(Duration.ofMillis(10L)))
> .aggregate(this::initialize, this::aggregateFields,
> materializedAsWindowStore(windowedStoreName,
> stringSerde,
> AggregateMetricsFieldsSerde))
>
> .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())
> .withName(windowedSuppressNodeName))
> .toStream().map((key, aggregateMetrics) ->
> { return KeyValue.pair(key.key(), aggregateMetrics); }
> );
> aggregatedData.to(aggregatedDataTopic, aggregateDataProduced);
> return aggregatedFlowData;
> }
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
