我正在使用帶有 Tumbling Window 的 Kafka Streams，然后是聚合步驟。但是觀察發(fā)出到聚合函數(shù)的元組數(shù)量正在下降。知道我哪里出錯(cuò)了嗎？代碼：  Properties props = new Properties();  props.put(StreamsConfig.APPLICATION_ID_CONFIG, "events_streams_local");  props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");  props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());  props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());  props.put(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, Arrays.asList(JmxReporter.class));  props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams/data/");  props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 20);  props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 60000);  props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, EventTimeExtractor.class);  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");  final StreamsBuilder builder = new StreamsBuilder();  HashGenerator hashGenerator = new HashGenerator(1);  builder  .stream(inputTopics)  .mapValues((key, value) -> {    stats.incrInputRecords();    Event event = jsonUtil.fromJson((String) value, Event.class);    return event;  })  .filter(new UnifiedGAPingEventFilter(stats))  .selectKey(new KeyValueMapper<Object, Event, String>() {    @Override    public String apply(Object key, Event event) {      return (String) key;    }  })  .groupByKey(Grouped.with(Serdes.String(), eventSerdes))  .windowedBy(TimeWindows.of(Duration.ofSeconds(30)))  .aggregate(new AggregateInitializer(), new UserStreamAggregator(), Materialized.with(Serdes.String(), aggrSerdes))  .mapValues((k, v) -> {    // update counter for aggregate records    return v;  })  .toStream()  .map(new RedisSink(stats));  topology = builder.build();  streams = new KafkaStreams(topology, props);每秒的 Redis 操作只是向下滑動(dòng)。