[jira] [Created] (KAFKA-13290) My timeWindows last aggregated message never emit until a new message coming

["Steve Zhou (Jira)" <ji...@apache.org>]

Steve Zhou created KAFKA-13290:
----------------------------------

             Summary: 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


I have a stream code which aggregate 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, even the new message has different key.

 

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(windowedFlowSuppressNodeName))
            .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)