You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Matthias J. Sax (Jira)" <ji...@apache.org> on 2021/09/10 22:14:00 UTC
[jira] [Commented] (KAFKA-13290) My timeWindows last aggregated
message never emit until a new message coming
[ 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)