我们有以下问题:
我们想听某个 Kafka 主题并构建它的“历史” - 因此对于指定的键提取一些数据，将其添加到该键的现有列表中(如果不存在则创建一个新列表)并将其放入另一个主题，它只有一个分区并且高度压缩。另一个应用程序可以只收听该主题并更新其历史记录列表。
我在想它如何与 Kafka 流库相适应。我们当然可以使用聚合:
msgReceived.map((key, word) -> new KeyValue<>(key, word))
.groupBy((k,v) -> k, stringSerde, stringSerde)
.aggregate(String::new,
(k, v, stockTransactionCollector) -> stockTransactionCollector + "|" + v,
stringSerde, "summaries2")
.to(stringSerde, stringSerde, "transaction-summary50");
它创建了一个由 Kafka 支持的本地存储并将其用作历史表。
我担心的是，如果我们决定扩展这样的应用程序，每个正在运行的实例都会创建一个新的支持主题
${applicationId}-${storeName}-changelog (我假设每个应用程序都有不同的 applicationId )。每个实例开始使用输入主题，获取一组不同的键并构建不同的状态子集。如果 Kafka 决定重新平衡，一些实例将开始错过本地存储中的一些历史状态，因为它们获得了一组全新的分区来消费。问题是，如果我只是为每个正在运行的实例设置相同的 applicationId，它最终是否应该重放来自每个正在运行的实例具有相同本地状态的相同 kafka 主题的所有数据？
最佳答案
为什么要创建多个具有不同 ID 的应用程序来执行相同的工作？ Kafka 实现并行的方式是通过任务:
An application’s processor topology is scaled by breaking it into multiple tasks.
More specifically, Kafka Streams creates a fixed number of tasks based on the input stream partitions for the application, with each task assigned a list of partitions from the input streams (i.e., Kafka topics). The assignment of partitions to tasks never changes so that each task is a fixed unit of parallelism of the application.
Tasks can then instantiate their own processor topology based on the assigned partitions; they also maintain a buffer for each of its assigned partitions and process messages one-at-a-time from these record buffers. As a result stream tasks can be processed independently and in parallel without manual intervention.
如果您需要扩展您的应用程序，您可以启动运行相同应用程序(相同应用程序 ID)的新实例，并且一些已经分配的任务将重新分配给新实例。本地状态存储的迁移将由库自动处理:
When the re-assignment occurs, some partitions – and hence their corresponding tasks including any local state stores – will be “migrated” from the existing threads to the newly added threads. As a result, Kafka Streams has effectively rebalanced the workload among instances of the application at the granularity of Kafka topic partitions.
我建议你看看this guide .
关于apache-kafka - Kafka Streams - 所有实例本地存储指向同一主题，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/41934043/