Kafka KStream through() 方法的使用 - 了解实例到分区的分配答案

【问题标题】：Use of the Kafka KStream through() method - Understanding the assignment of instances to partitionsKafka KStream through() 方法的使用 - 了解实例到分区的分配
【发布时间】：2019-02-23 14:27:14
【问题描述】：

我在使用 KStream through() 方法来确保将消息分发到正确的分区时遇到了一些麻烦。

这里有一些背景。我有一个 kafka 流应用程序，它在 inTopic 上侦听 CustomerEvent 并在 outTopic 上写入 CatalogEvent：

inTopic ---> MY_KAFKA_STREAM_APPLICATION ---> outTopic

在inTopic 上，键和值是(AccountId, CustomerEvent)。
在outTopic上，key和value分别是(CatalogId, CatalogEvent)

我正在使用 KStream transform() 方法将 CustomerEvent 转换为 CatalogEvent。我需要使用transform()，因为CatalogEvent 依赖于以前的CustomerEvents 共享我过去看到的相同CatalogId，因此将涉及到一个状态存储。

这是我初始化状态存储的方式。我将通过 CatalogId 查询状态存储，以检索有关以前共享相同 CatalogId 的以前 CustomerEvents 的信息。

StoreBuilder<KeyValueStore<String, MyAggregator>> catalogStore = 
     Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myStore"), Serdes.String(), aggregatorSerde)
           .withLoggingEnabled(new HashMap<>());

builder.addStateStore(catalogStore);

这是我设置拓扑的方式：

builder.stream("inTopic", Consumed.with(Serdes.String(), customerEventSerde))
    .selectKey((k, customerEvent) -> customerEvent.getCatalogId())
    .through("bycatalogid", Produced.with(Serdes.String(), customerEventSerde))
    .transform(()-> new MyTransformer("myStore"), "myStore")
    .to("outTopic", Produced.with(Serdes.String(), catalogEventSerde));

我需要确保共享同一个 CatalogId 的所有 CustomerEvents 最终都位于同一个分区上。这就是我使用selectKey() 将密钥从AccountId 更改为CatalogId 并使用through() 方法的原因。

我正在为我的主题使用 2 个分区、我的 kafka 流应用程序的 2 个实例和 1 个 kafka 服务器进行测试。

我正在使用以下命令来查看我的实例是如何分配给每个分区的：

优秀的作业案例

kafka-consumer-groups.sh --describe --group my_application_group --bootstrap-server 192.168.92.118:9092
Note: This will not show information about old Zookeeper-based consumers.


TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG CONSUMER-ID
bycatalogid    1          -               0               -   consumer1/192.168.92.118 
inTopic        1          9               9               0   consumer1/192.168.92.118
bycatalogid    0          5               5               0   consumer2/192.168.92.29
inTopic        0          12              12              0   consumer2/192.168.92.29

有时分配很好，如上所示。所有消费者都被正确分配。在 192.168.92.118 上运行的实例被分配到 partition1，在 192.168.92.29 上运行的实例被分配到 partition0。此外，我可以看到共享相同 CatalogId 的所有 CustomerEvents 都被发送到同一个分区。

但是，有时当我重新启动实例时，将实例分配给分区是错误的：

错误的分配案例

kafka-consumer-groups.sh --describe --group my_application_group --bootstrap-server 192.168.92.118:9092
Note: This will not show information about old Zookeeper-based consumers.

TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG CONSUMER-ID
bycatalogid    0          11              11              0   consumer1/192.168.92.118
bycatalogid    1          3               3               0   consumer1/192.168.92.118
inTopic        0          18              18              0   consumer2/192.168.92.29
inTopic        1          12              12              0   consumer2/192.168.92.29

上面的赋值完全没有意义。运行在 192.168.92.118 上的实例仅监听主题 bycatalogid，而 192.168.92.29 上的另一个实例仅监听主题 inTopic。这怎么可能？

另外，出于调试目的，我在我的服务中实现了一个 REST api，我可以发送一个带有 CatalogId 的 HTTP GET 请求来检索我的 kafka 存储中的内容。我正在像这样访问我的 kafka 商店：

ReadOnlyKeyValueStore<String, MyAggregator> catalogStore 
    = streams.store("myStore", QueryableStoreTypes.<String, MyAggregator>keyValueStore());

如果在仅在 inTopic 分区上侦听的实例上执行上述操作，则会引发以下异常。

Caused by: org.apache.kafka.streams.errors.InvalidStateStoreException: The state store, myStore, may have migrated to another instance.

我需要做些什么来确保我不会收到BAD ASSIGNMENT CASE。

谢谢。

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

这两种情况都有效。您的输入主题和通过主题都有 2 个分区。因此，创建了 4 个任务，每个任务都是一个独立的处理单元，可以独立于所有其他任务分配。

没有理由一个实例需要处理来自两个不同主题的分区。

对于交互式查询：以上就是商店可以托管在不同实例上的原因。（参见https://kafka.apache.org/21/documentation/streams/developer-guide/interactive-queries.html）

【讨论】：