Кафка потоковые записи не пересылаются после создания окон / агрегации

Question

Я использую Kafka Streams с Tumbling Window с последующим шагом агрегирования. Но наблюдение за числом кортежей, испускаемых для агрегатной функции, уменьшается. Любая идея, где я иду не так?

Код:

  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 операций в секунду просто скользит вниз.

Answer 1

Kafka Streams использует кэши в хранилище состояний, чтобы уменьшить нагрузку на нисходящий поток. Если вы хотите получать каждое обновление магазина в виде нисходящей записи, вы можете установить нулевой размер кэша с помощью StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG (глобально для всех магазинов) или для каждого магазина с помощью передачи Materialized.as(...).withCachingDisabled() соответствующему оператору (например, * 1003). *).

Проверьте документы для более подробной информации: https://docs.confluent.io/current/streams/developer-guide/memory-mgmt.html