apache-kafka - 如果第一个代理宕机，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

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

关闭代理实例后消费者的日志:

[2018-01-09 13:39:31,130] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 2147483646 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,132] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,344] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 2147483646 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,451] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,848] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 2147483646 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:31,950] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:32,363] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 2147483646 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:33,092] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 2147483646 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:34,216] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 2147483646 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient) [2018-01-09 13:39:34,218] WARN [Consumer clientId=consumer-1, groupId=test] Asynchronous auto-commit of offsets {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=''}} failed: Offset commit failed with a retriable exception. You should retry committing offsets. The underlying error was: The coordinator is not available. (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:39:34,219] WARN [Consumer clientId=consumer-1, groupId=test] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

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

[2018-01-09 13:41:21,739] ERROR [Consumer clientId=consumer-1, groupId=test] Offset commit failed on partition demo-0 at offset 3: This is not the correct coordinator. (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:41:21,739] WARN [Consumer clientId=consumer-1, groupId=test] Asynchronous auto-commit of offsets {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=''}} failed: Offset commit failed with a retriable exception. You should retry committing offsets. The underlying error was: This is not the correct coordinator. (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:41:22,353] ERROR [Consumer clientId=consumer-1, groupId=test] Offset commit failed on partition demo-0 at offset 3: This is not the correct coordinator. (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator) [2018-01-09 13:41:22,354] WARN [Consumer clientId=consumer-1, groupId=test] Asynchronous auto-commit of offsets {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=''}} failed: Offset commit failed with a retriable exception. You should retry committing offsets. The underlying error was: This is not the correct coordinator. (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