Kafka-Streams：为什么所有分区都分配给ConsumerGroup中的同一个使用者？

Question

多台机器生成事件。这些事件被发送到我们的Kafka集群，其中每台机器都有自己的主题（app.machine-events.machine-name）。因为顺序在每台机器的基础上很重要，而分区大小现在不是问题，所以所有主题都由一个分区组成。因此，目前，N个主题也意味着N个分区。

消费/处理应用程序使用了kafka-streams，我们给出了streamsconfig.application_id_config/“application.id”“machine-event-processor”，它对每个实例都是相同的，这意味着它们会被放入Kafka的同一个使用者组中。这个使用者订阅了app.machine-events.*模式，至于处理器，它处理哪台机器的事件并不重要。通过./kafka-consumer-groups.sh--bootstrap-server localhost:9092-description-group machine-event-processor--members--verbose验证了这一点，它向我显示了一个与正在运行的所有处理服务的&IP数量匹配的列表。

预期

给定20台机器和5个处理器实例，我们希望每个处理器处理4个分区（因此处理4个主题）。

实际上

一个处理器处理20个分区（因此20个主题），其他4个处理器什么都不做/空闲。杀死“幸运”处理器，所有20个分区都被重新平衡到另一个处理器，导致新处理器处理20个分区/主题，并且3个处理器空闲。

到目前为止我所尝试的

• 检查partition.grouper。我觉得我不是完全理解它，但就我所能找到的，反正只有DefaultPartitioner选项，编写一个自定义选项应该是不必要的，因为（根据文档）这个设置应该可以工作。它确实提到了分区会根据它们的分区键（对于我们来说都是0，因为每个主题只有一个分区）加入到任务中，但我不能完全理解这一部分。
• 已将RoundRobinAssignor用于使用者:设置.put(StreamsConfig.ConsumerPrefix(ConsumerConfig.Partition_Assignment_Strategy_Config),new RoundRobinAssignor().GetClass.GetName)（尝试了几个值，因为似乎没有任何变化。）
• 检查其他配置属性，看看是否遗漏了什么：我想没有。

简化的代码

val streamConfig = new Properties
// {producer.metadata.max.age.ms=5000, consumer.metadata.max.age.ms=5000, default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde, consumer.partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor, bootstrap.servers=kafka:9092, application.id=machine-event-processor, default.value.serde=org.apache.kafka.common.serialization.Serdes$ByteArraySerde}
val builder: StreamsBuilder = new StreamsBuilder
val topicStream: KStream[String, Array[Byte]] = builder.stream(Pattern.compile("app.machine-events.*"))
topicStream.process(new MessageProcessorSupplier(context)) // The event is delegated to a processor, doing the actual processing logic
val eventStreams = new KafkaStreams(builder.build(), streamConfig)
eventStreams.start()

备注

>

正在使用Kafka-Streams 2.0.0:

org.apache.kafka kafka-streams 2.0.0

Kafka正在容器中运行，使用Wurstmeister/Kafka:2.11-2.0.0版本。docker-stack.yml服务：

<代码>kafka：image：wurstmeister/kafka：2.11-2.0.0端口：-目标：9094发布：9094协议：tcp模式：主机卷：-/var/run/docker.sock：/var/run/docker.sock健康检查：测试：[“cmd-shell”,“$$(netstat-ltn grep-q 9092）”]间隔：15s超时：10s重试：5环境：HOSTNAME_COMMAND：“docker info grep^名称：cut-d''-f 2”kafka_zookeeper_connect:zookeeper：2181

• Kafka设置在双节点设置中，形成集群。通过docker环境变量，我们将replication因子设置为2，因此每个分区在每个节点上都应该有一个复制。

我找到并检查的相关主题/问题/讨论

>

KIP-49

https://faust.readthedocs.io/en/latest/developerguide/partition_assignor.html

我查了Kafka的邮件档案，但没有找到任何东西

签出的流示例应用程序

全方位地寻找别人遇到的这个问题，但没有给我我需要的答案。我还发现了KAFKA-7144，但这对我们来说不应该是一个问题，因为我们正在运行2.0.0

如果有人遇到过类似的问题，或者能指出我可能很愚蠢的错误，请指点我！

Answer 1

对于将来遇到同样问题的读者，解决方案是不使用每个有1个分区的N个主题，而是使用有N个分区的1个主题。即使有120个分区和400+个机器/事件源，多个事件类型也会被放入同一个分区，但这不会影响事件的顺序。

实现方法是将记录键设置为机器名，并让底层逻辑负责哪个值进入哪个分区。由于我们现在有一个消费者组，其中有X个消费者订阅了本主题，因此分区被平均分配给消费者，每个消费者使用120/X个分区。

这正如Matthias所建议的，在Devoxx Belgium 2018展会上，来自Confluent的其他乐于助人的人士进一步证实了这一点。谢谢你！

小费

使用WurstMeister/Kafka docker映像时,请考虑使用environment属性：

kafka_create_topics:“app.machine-events：120：2”

意义

主题名称：分区数：复制因子