如果第一个代理关闭，Kafka 消费者无法消费

Question

我正在使用最新版本的 kafka(kafka_2.12-1.0.0.tgz)。我已经设置了带有 3 个代理的简单集群（只是在每个实例的属性文件中更改了 broker.id=1 和 listeners=PLAINTEXT://:9092）。集群启动后，我使用以下命令创建了主题

./kafka-topics.sh --create    --zookeeper localhost:2181  --replication-factor 3     --partitions 13    --topic demo

然后使用以下命令启动 kafka 消费者和生产者

./kafka-console-producer.sh --topic  demo  --broker-list localhost:9094,localhost:9093,localhost:9092

./kafka-console-consumer.sh --group test --bootstrap-server localhost:9094,localhost:9093,localhost:9092  --topic demo

当所有经纪人都起来时，一切都很好。但是，如果我首先杀死（按启动顺序）代理消息被发送到代理但消费者无法接收任何消息。消息不会丢失。启动该代理消费者后立即收到消息。

关闭broker实例后的消费者日志：

[2018-01-09 13:39:31,130] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 2147483646 的连接。 经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,132] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 1 的连接。经纪人 可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,344] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 2147483646 的连接。 经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,451] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 1 的连接。经纪人 可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,848] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 2147483646 的连接。 经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,950] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 1 的连接。经纪人 可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:32,363] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 2147483646 的连接。 经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:33,092] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 2147483646 的连接。 经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:34,216] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 2147483646 的连接。 经纪人可能不可用。 (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:34,218] 警告 [消费者 clientId=consumer-1， groupId=test] 异步自动提交偏移量 {demo-0=OffsetAndMetadata{offset=3, metadata=''}, demo-1=OffsetAndMetadata{offset=3, metadata=''}, demo-2=OffsetAndMetadata{offset=2, metadata=''}, demo-3=OffsetAndMetadata{offset=2, metadata=''}, demo-4=OffsetAndMetadata{offset=1, metadata=''}, demo-5=OffsetAndMetadata{offset=1, metadata=''}, demo-6=OffsetAndMetadata{offset=3, metadata=''}, demo-7=OffsetAndMetadata{offset=2, metadata=''}, demo-8=OffsetAndMetadata{offset=3, metadata=''}, demo-9=OffsetAndMetadata{offset=2, metadata=''}, demo-10=OffsetAndMetadata{offset=3, metadata=''}, demo-11=OffsetAndMetadata{offset=2, metadata=''}, demo-12=OffsetAndMetadata{offset=2, metadata=''}} 失败：偏移 提交失败，出现可重试异常。您应该重试提交 偏移量。潜在的错误是：协调器不可用。 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:39:34,219] 警告 [消费者 clientId=consumer-1， groupId=test] 无法建立到节点 1 的连接。经纪人 可能不可用。 (org.apache.kafka.clients.NetworkClient)

再次启动缺少代理后的消费者日志：

[2018-01-09 13:41:21,739] 错误 [Consumer clientId=consumer-1, groupId=test] 在偏移量 3 的分区 demo-0 上的偏移量提交失败： 这不是正确的协调员。 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:41:21,739] 警告 [消费者 clientId=consumer-1， groupId=test] 异步自动提交偏移量 {demo-0=OffsetAndMetadata{offset=3, metadata=''}, demo-1=OffsetAndMetadata{offset=3, metadata=''}, demo-2=OffsetAndMetadata{offset=2, metadata=''}, demo-3=OffsetAndMetadata{offset=2, metadata=''}, demo-4=OffsetAndMetadata{offset=1, metadata=''}, demo-5=OffsetAndMetadata{offset=1, metadata=''}, demo-6=OffsetAndMetadata{offset=3, metadata=''}, demo-7=OffsetAndMetadata{offset=2, metadata=''}, demo-8=OffsetAndMetadata{offset=3, metadata=''}, demo-9=OffsetAndMetadata{offset=2, metadata=''}, demo-10=OffsetAndMetadata{offset=3, metadata=''}, demo-11=OffsetAndMetadata{offset=2, metadata=''}, demo-12=OffsetAndMetadata{offset=2, metadata=''}} 失败：偏移 提交失败，出现可重试异常。您应该重试提交 偏移量。潜在的错误是：这不是正确的 协调员。 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:41:22,353] 错误 [消费者 clientId=consumer-1， groupId=test] 在偏移量 3 的分区 demo-0 上的偏移量提交失败： 这不是正确的协调员。 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:41:22,354] 警告 [消费者 clientId=consumer-1， groupId=test] 异步自动提交偏移量 {demo-0=OffsetAndMetadata{offset=3, metadata=''}, demo-1=OffsetAndMetadata{offset=3, metadata=''}, demo-2=OffsetAndMetadata{offset=2, metadata=''}, demo-3=OffsetAndMetadata{offset=2, metadata=''}, demo-4=OffsetAndMetadata{offset=1, metadata=''}, demo-5=OffsetAndMetadata{offset=1, metadata=''}, demo-6=OffsetAndMetadata{offset=3, metadata=''}, demo-7=OffsetAndMetadata{offset=2, metadata=''}, demo-8=OffsetAndMetadata{offset=3, metadata=''}, demo-9=OffsetAndMetadata{offset=2, metadata=''}, demo-10=OffsetAndMetadata{offset=3, metadata=''}, demo-11=OffsetAndMetadata{offset=3, metadata=''}, demo-12=OffsetAndMetadata{offset=2, metadata=''}} 失败：偏移 提交失败，出现可重试异常。您应该重试提交 偏移量。潜在的错误是：这不是正确的 协调员。 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)

谢谢

Answer 1

尝试检查 server-*.properties 文件中的“offsets.topic.replication.factor”

例如：

############################# Internal Topic Settings       
# The replication factor for the group metadata internal topics    
# For anything other than development testing, a value greater than 1 is  recommended for to ensure availability such as 3.
offsets.topic.replication.factor=3

http://kafka.apache.org/documentation/#brokerconfigs

Answer 2

在yml 文件上使用KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR 可解决此问题。

例如在 docker-swarm 上使用 2 个工作人员。

environment:
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2