重启后是否可以在不使用更改日志主题的情况下恢复 Kafka Streams 状态存储？

Question

我们有 2 个 compacted 主题，每个主题都包含数 TB 的数据，我们想使用 Spring Cloud Stream 和 Kafka Streams 加入它们。 （简化的）代码如下所示：

@Bean
public BiConsumer<KTable<String, LeftEvent>, KTable<String, RightEvent>> processEvents() {
    return ((leftEvents, rightEvents) -> {
      leftEvents.join(rightEvents, this::merge)
        .toStream()
        .foreach(this::process);
    });
}

这种方法的问题在于，使用 KTables 作为输入参数会导致创建 changelog 主题，这实际上是 duplicate[=41] =] 源主题，因为如上所述，这两个主题都已经压缩。为了避免在 Kafka 中复制数 TB 的数据，我们的第一个尝试是使用 KStreams 作为输入，并将它们转换为 KTables，如下所示：

stream.toTable(
  Materialized
    .<K, V, KeyValueStore<Bytes, byte[]>>as(stateStoreName)
    .withLoggingDisabled()
);

从而禁用日志记录并因此免除更改日志主题，这在我们的上下文中似乎没有用。

但是，以下情况现在不再适用：

1. 用键 k1
生成一个 LeftEvent
2. 重新启动应用程序
3. 用键 k1
生成一个 RightEvent

事件不再加入，但如果应用程序未在中间重新启动（即第 1 步，然后第 3 步），加入工作正常。

当应用程序重新启动时，我们希望在没有更改日志主题的情况下从源主题重建状态存储，但事实显然并非如此。在某些情况下，我们观察到 rocksDB 文件（位于 /tmp/kafka-streams/...）用于检索重启前消耗的数据，但是我们不能假设这些文件在重启后仍然可用，因为我们在容器化中工作环境。

有没有一种方法可以支持重启（并实现容错）而不必使用更改日志主题，在我们的例子中，它会复制输入主题？如果不是，我们可能不得不重新考虑我们对 Kafka Streams 的使用...

Answer 1

您想启用 Kafka 流的优化：https://docs.confluent.io/platform/current/streams/developer-guide/optimizing-streams.html#optimization-details（#1 是您要找的）。

目前，Kafka Streams 在启用时执行了两项优化：

1. 源 KTable 重新使用源主题作为变更日志主题。
2. 如果可能，Kafka Streams 会将多个重新分区主题合并为一个重新分区主题。

要指出的关键是，由于我自己犯了这个错误，所以不要忘记将配置发送到 build() 和 KStreams 的构造（优化，如 link 提供）在构建中完成。

// tell Kafka Streams to optimize the topology
config.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION, StreamsConfig.OPTIMIZE);

// Since we've configured Streams to use optimizations, the topology is optimized during the build.
// And because optimizations are enabled, the resulting topology will no longer need to perform
// three explicit repartitioning steps, but only one.
final Topology topology = builder.build(config);
final KafkaStreams streams = new KafkaStreams(topology, config);

现在所有拓扑都启用了优化，因此如果您记住这一点，还会执行 #2 优化。