даю данные обработки KTable из KStream. Но когда я запускаю сообщения-захоронения с ключом и нулевой полезной нагрузкой, это не удаляет сообщение из KTable.

образец -

public KStream<String, GenericRecord> processRecord(@Input(Channel.TEST) KStream<GenericRecord, GenericRecord> testStream,
KTable<String, GenericRecord> table = testStream
                .map((genericRecord, genericRecord2) -> KeyValue.pair(genericRecord.get("field1") + "", genericRecord2))
                .groupByKey()
                reduce((genericRecord, v1) -> v1, Materialized.as("test-store"));


GenericRecord genericRecord = new GenericData.Record(getAvroSchema(keySchema));
genericRecord.put("field1", Long.parseLong(test.getField1()));
ProducerRecord record = new ProducerRecord(Channel.TEST, genericRecord, null);
kafkaTemplate.send(record);

После запуска сообщения с нулевым значением я могу отладить в функции карты testStream с нулевой полезной нагрузкой, но она не удаляет запись в журнале изменений KTable «test-store». Похоже, что он даже не достигает метода снижения, не уверен, что мне здесь не хватает.

Цени любую помощь в этом!

Благодарю.