我们有一个KStream应用程序，它使用内存KVStateStore，但禁用了更新日志。

  String stateStoreName = "statestore-v1";
  StoreBuilder<KeyValueStore<String, Event>> keyValueStoreBuilder =
      Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore(stateStoreName), 
          Serdes.String(), new JsonSerde<>(Event.class));
  keyValueStoreBuilder.withLoggingDisabled();
  streamsBuilder.addStateStore(keyValueStoreBuilder);

我们现在想启用具有不同配置和不同名称的更新日志。

  String stateStoreName = "statestore-v2";
  StoreBuilder<KeyValueStore<String, Event>> keyValueStoreBuilder =
      Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore(stateStoreName), 
          Serdes.String(), new JsonSerde<>(Event.class));
  Map<String, String> changelogConfig = new HashMap<>();
  changelogConfig.put("retention.ms", "43200000"); // 12 hours
  changelogConfig.put("cleanup.policy", "delete");
  changelogConfig.put("auto.offset.reset", "latest");
  keyValueStoreBuilder.withLoggingEnabled(changelogConfig);
  streamsBuilder.addStateStore(keyValueStoreBuilder);

当我们运行我们的应用程序时，我们进入了这些消息的无限循环：

2022-10-11 13:02:32.761 app=myapp INFO 54561 --- [-StreamThread-3]
o.a.k.s.p.i.StoreChangelogReader : stream-thread [myapp-StreamThread-3] 
End offset for changelog myapp-statestore-v2-changelog-4 cannot be found; 
will retry in the next time.
2022-10-11 13:02:32.761 app=myapp INFO 54561 --- [-StreamThread-3] 
o.a.k.clients.consumer.KafkaConsumer : [Consumer clientId=myapp-StreamThread-3-restore-consumer, groupId=null] 
Unsubscribed all topics or patterns and assigned partitions

似乎从未创建过更新日志主题…至少kafka-topic没有显示它。

我正在使用io. confluent软件包版本7.2.2-ccs，我认为它可以转换为Apache Kafka版本3.2.x

关于如何修复无限循环并创建更新日志主题的任何想法？

谢谢！