Kafka Streaming任务和内部状态存储的管理

如何解决Kafka Streaming任务和内部状态存储的管理

让我们说我们已经在具有以下属性的2台不同机器（实例）上启动了2个Streaming-Tasks：-

public final static String applicationID = "StreamsPOC";
public final static String bootstrapServers = "10.21.22.56:9093";    
public final static String topicname = "TestTransaction";
public final static String shipmentTopicName = "TestShipment";
public final static String RECORD_COUNT_STORE_NAME = "ProcessorONEStore";

并使用上述这些属性，以下是流任务的定义：-

        Map<String,String> changelogConfig = new HashMap();
        changelogConfig.put("min.insyc.replicas","1");
        // Below line not working.
        changelogConfig.put("topic","myChangedTopicLog");
       
        StoreBuilder kvStoreBuilder = Stores.keyValueStoreBuilder(
                Stores.persistentKeyValueStore(AppConfigs.RECORD_COUNT_STORE_NAME),AppSerdes.String(),AppSerdes.Integer()
        ).withLoggingEnabled(changelogConfig);

        kStreamBuilder.addStateStore(kvStoreBuilder);


        KStream<String,String> sourceKafkaStream = kStreamBuilder.stream
                (AppConfigs.topicname,Consumed.with(AppSerdes.String(),AppSerdes.String()));

现在，正如我观察到的那样，卡夫卡在幕后创建了一个主题（为了备份内部状态存储），其名称如下：-StreamsPOC-ProcessorONEStore-changelog

第一个问题是：-是否两个不同的流任务都维护内部状态存储并将其备份到同一主题？

第二个问题是-假设Task-1在分区1上启动，并且在其本地内部状态存储中写入“ ”，然后Task-2开始在Partition-2上工作并说还将写入其各自的本地状态存储中，那么这是否不会引发数据被覆盖的风险，因为这两个任务都将数据备份到相同的changelog主题？

第三个问题是：-如何将自定义名称指定为Change-log-topic？

应高度赞赏响应！

解决方法

首先，对术语有一些思考：“任务”一词在Kafka Stream中具有明确定义的含义，作为用户，您不能自己创建任务。执行程序时，Kafka Streams将创建“任务”，它们是“独立于计算单位”，并为您执行这些任务。 -我猜，“任务”的意思实际上是一个KafkaStreams客户端（称为 instance ）。

如果您使用相同的application.id启动多个实例，它们将组成一个使用者组，并且它们将以并行数据的方式共享负载。对于状态存储，每个实例都将托管该存储的 shard （有时也称为分区）。所有实例都使用相同的主题，并且该主题为每个存储分片都有一个分区。从存储分片到更改日志分区有1：1映射。此外，从输入主题分区到任务有1：1映射，在任务和存储碎片之间有1：1映射。因此，总体而言，这是1：1：1：1映射：对于每个输入主题分区，将创建一个任务，并且每个任务都保存一个状态存储区的碎片，每个存储区碎片由changelog主题的一个分区支持。 （最重要的是，输入主题分区的数量决定了您获得多少个并行任务和存储分片，并且changelog主题的创建数量与输入主题的分区数量相同。）

1. 是的，所有实例都使用相同的changelog主题。
2. 由于任务是通过分片和changelog主题分区隔离的，因此它们不会相互覆盖。但是，任务的思想是每个任务处理不同的（不重叠）键空间，因此，具有相同<k1,...> 的所有记录应由相同的任务处理。当然，该规则可能会有例外，如果您的应用程序不使用非重叠键空间，则程序将仅被执行（当然，根据您的业务逻辑要求，这可能是正确的还是不正确的。）
3. 似乎您已经做过：请注意，您只能自定义变更日志主题名称的一部分：<application.id>-<storeName>-changelog-即，您可以选择application.id和storeName。不过，整体命名模式是硬编码的。