在一个实例发生故障转移时，驻留在kafka流实例中的本地状态存储未同步

Question

我在本地机器中运行多个kafka流消费者实例（2个实例），每个实例都有自己的自定义本地存储，每个实例的名称不同。

根据文档，如果其中一个实例发生故障，则kafka必须将死实例的存储同步到活实例的存储（如果我错了，请更正我）。

我用相同的应用程序id配置了两个实例，让kafka知道这些实例属于同一个组。

当其中一个实例被杀死时，另一个（活动）实例的存储未与死实例的存储同步。我在两个商店都启用了更改日志主题。

然而，当我在两个实例中都有相同的商店名称时，商店会按预期同步，不确定这些实例是否指向一个商店。我有不同的StreamsConfig。这两个实例的STATE\u DIR\u CONFIG位置。

请让我知道，如果我遗漏了什么，存储名称可以在不同的应用程序实例上不同吗？kafka是否会自动在新实例存储上重播更改日志主题？

//下面是我的流配置

@Bean
public KafkaStreams kafkaStreams(KafkaProperties properties,
                                 @Value("${spring.application.name}") String appName) {
    final Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, appName);
    props.put(StreamsConfig.CLIENT_ID_CONFIG, "client2");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
    props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class);
    props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
    //props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams1");
    props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, "1");
    props.put(StreamsConfig.consumerPrefix(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG),
            new RoundRobinAssignor().getClass().getName());
    props.put("auto.offset.reset", "earliest");
    final KafkaStreams kafkaStreams = new KafkaStreams(kafkaStreamTopology(), props);
    System.out.println("Invoked kafkaStreams");
    //kafkaStreams.cleanUp();
    kafkaStreams.start();
    return kafkaStreams;
}

Answer 1

我在本地机器中运行多个kafka流消费者实例（2个实例），每个实例都有自己的自定义本地存储，每个实例的名称不同。

这听起来不正确。如果您使用相同的application.id（即group.id）运行多个实例，则所有实例都必须执行相同的代码。（我想知道为什么您的应用程序一开始就没有崩溃。）

我不能100%确定你想实现什么。如果你能分享你的拓扑代码，这可能会有所帮助？

请注意，KafkaStreams基于进审量主题分区（参见https://docs.confluent.io/current/streams/architecture.html）对逻辑存储进行分片。也许您混淆了分片和逻辑存储？

如果您想拥有两个逻辑存储，每个存储有一个分片，您仍然可以运行多个实例，并且存储将在不同的实例上执行（故障转移也可以工作）。但是，您仍然需要在启动时在两个实例上“包含”两个存储。