Kafka Stream Suppress在窗口化后不产生输出

Question

我正在从流环境中打开进入的对象的窗口，对其进行收集和打印。使用kafka抑制可避免出现中间结果。

使用抑制后，我将无法输出任何信息。如果我注释掉抑制代码，则代码可以正常工作，但会输出中间结果。

import com.savk.workout.kafka.streams.kafkastreamsworkout.config.ConfigUtils;
import com.savk.workout.kafka.streams.kafkastreamsworkout.model.Observation;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Suppressed;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.kafka.support.serializer.JsonSerde;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Properties;

@Component
public class ObservationAnalyser {

    @PostConstruct
    public void initialize() {
        StreamsBuilder streamsBuilder;
        KStream<String, Observation> observationKStream;

        String observationSerde = ConfigUtils.getObservationSerde();        //TODO : Should we move to a JSON Serde?
        Properties kafkaStreamProperties = ConfigUtils.getKafkaStreamConfig();
        kafkaStreamProperties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        kafkaStreamProperties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, observationSerde);

        //JsonSerde<ObservationCollector> observationCollectorJsonSerde = new JsonSerde<>(ObservationCollector.class);

        streamsBuilder = new StreamsBuilder();
        observationKStream = streamsBuilder.stream(ConfigUtils.KAFKA_SOURCE_TOPIC);
        observationKStream
                .groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofMillis(ConfigUtils.ONE_MINUTE_IN_MILLISECONDS)))
                .aggregate(
                        () -> new ObservationCollector(),
                        (key, value, observationCollector) -> observationCollector.addObservations(value),
                        Materialized.with(Serdes.String(), new JsonSerde<>(ObservationCollector.class))
                )
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))    //AFTER COMMENTING THIS LINE, I CAN SEE THE OUTPUT
                .toStream((key, value) -> key.key())
                .foreach((key, observationCollector) -> {
                    System.out.println("Key :: " + key);
                    for(Observation observation : observationCollector.getObservations())   {
                        System.out.println("Observation :: " + observation);
                    }
                });

        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), kafkaStreamProperties);
        kafkaStreams.start();
        Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
    }

}

我无法找出问题/问题或寻找任何资源进行故障排除。

Answer 1

默认的“宽限期”为1天（如果不使用suppress()，则出于向后兼容的原因）。因此，在事件时间提前1天之前，该窗口不会关闭。

您可能希望通过以下方式减少宽限期（也许还有保留时间）>>

• TimeWindows.of(..).grace(...)
• Materialized.with(...).withRetentionTime(...)