如何清理Kafka KStream内部变更日志和重新分区主题

Question

kafka版本1.1

我们使用Kafka KStream根据事件本身中选定的键聚合事件。大致低于它的作用

KStream[String, Event]
.selectKey[String]({ (_, event) =>
        aggregationKey(event)
      }
.groupByKey()
.aggregate(
        () => {
          Event("", "")
        },
        (k: Any, event: Event, aggregate: Event) => aggregator(k, event, aggregate, inputTopicName),
        Materialized
          .as[String, Event, KeyValueStore[Bytes, Array[Byte]]]("store-name")
          .withValueSerde(protoSerde)
      )
.toStream
.to(outTopicName)

在“聚合器”函数中，我在一定条件下返回空值，以产生墓碑事件。

kafka创建两个主题，重新分区和变更。在重新分区主题中保留设置为-1。不管墓碑事件如何，这些主题都在继续增长。我找不到清理它们的方法。

我们的要求是直截了当的：

只要满足密钥的条件，该密钥的聚合更改日志就没有用处。我们希望彻底永久地清除该密钥的所有事件

请建议如何清理kstream内部主题的基础上的关键？非常感谢。

Answer 1

上游重新分区主题不应该不受限制地增长：因为您注意到保留时间设置为-1（以避免数据丢失）。但是，在处理记录后，KafkaStreams会显式清除该主题。

此外，如果您从您的Aggregator返回null，相应的条目将在KTable存储中删除，墓碑将被发送到变更日志主题并发送到下游。当然，墓碑也首先附加到主题，只有当代理方主题压缩运行时，旧记录才会被“垃圾收集”。