kafka 消费者尝试连接到随机主机名而不是正确的一个答案

【问题标题】：kafka consumer try to connect to random hostname instead right onekafka 消费者尝试连接到随机主机名而不是正确的一个
【发布时间】：2018-07-07 16:37:01
【问题描述】：

我是 Kafka 的新手，并开始使用示例程序进行探索。它过去可以正常工作，但是突然 consumer.poll() 命令挂起并且永远不会返回。谷歌搜索建议检查服务器是否可以访问。 Producer 和 Consumer java 代码在同一台机器上运行，producer 可以将记录发布到 Kafka，但 consumer poll 方法挂起。

环境：
卡夫卡版本：1.1.0
客户端：Java
在 Windows 内的 Ubuntu docker 容器中运行
Zookeeper 和 2 个 Broker 服务器运行在同一个容器中

当我为客户端代码启用日志记录时，我看到以下异常：

2018-07-06 21:24:18 DEBUG NetworkClient:802 - [Consumer clientId=consumer-1, groupId=IDCS_Audit_Event_Consumer] Error connecting to node 4bdce773eb74:9095 (id: 2 rack: null)
java.io.IOException: Can't resolve address: 4bdce773eb74:9095
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:235)
    at org.apache.kafka.common.network.Selector.connect(Selector.java:214)
    .................
    .................

我不确定为什么消费者尝试连接到4bdce773eb74，即使我的代理服务器是192.168.99.100:9094,192.168.99.100:9095。还有我的完整消费者代码：

        final String BOOTSTRAP_SERVERS = "192.168.99.100:9094,192.168.99.100:9095";
        final Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "Event_Consumer");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props);
        TopicPartition tpLogin = new TopicPartition("login1", 0);
        TopicPartition tpLogout = new TopicPartition("logout1", 1);
        List<TopicPartition> tps = Arrays.asList(tpLogin, tpLogout);
        consumer.assign(tps);
        while (true) {
            final ConsumerRecords<Long, String> consumerRecords = consumer.poll(1000);
            if (consumerRecords.count()==0) {
                continue;
            }
            consumerRecords.forEach(record -> {
                System.out.printf("Consumer Record:(%d, %s, %d, %d)\n", record.key(), record.value(),
                        record.partition(), record.offset());
            });

            consumer.commitAsync();
            Thread.sleep(5000);
        }
    }

请帮助解决这个问题。

编辑正如我之前所说，我有 2 个经纪人，比如经纪人 1 和经纪人 2。如果我停止 broker-1，则不会记录上述异常，但 poll() 方法仍然没有返回。如果我停止 broker-1，以下消息将无限期记录：

2018-07-07 11:31:24 DEBUG AbstractCoordinator:579 - [Consumer clientId=consumer-1, groupId=IDCS_Audit_Event_Consumer] Sending FindCoordinator request to broker 192.168.99.100:9094 (id: 1 rack: null)
2018-07-07 11:31:24 DEBUG AbstractCoordinator:590 - [Consumer clientId=consumer-1, groupId=IDCS_Audit_Event_Consumer] Received FindCoordinator response ClientResponse(receivedTimeMs=1530943284196, latencyMs=2, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=1, clientId=consumer-1, correlationId=573), responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=COORDINATOR_NOT_AVAILABLE, node=:-1 (id: -1 rack: null)))
2018-07-07 11:31:24 DEBUG AbstractCoordinator:613 - [Consumer clientId=consumer-1, groupId=IDCS_Audit_Event_Consumer] Group coordinator lookup failed: The coordinator is not available.
2018-07-07 11:31:24 DEBUG AbstractCoordinator:227 - [Consumer clientId=consumer-1, groupId=IDCS_Audit_Event_Consumer] Coordinator discovery failed, refreshing metadata

提前致谢，索曼

【问题讨论】：

哪个版本的 Kafka？最新的是每个docs 的 1.1.0。 2.11 指的是 Scala 的版本。
@NickPredey 已经下载了kafka_2.11-1.1.0，那么我的Kafka版本是1.1.0？
对，就是那个版本。
你试过控制台消费者吗？ @用户
尝试个人经纪人@user

标签： java apache-kafka

【解决方案1】：

我发现了问题。当我创建主题时，broker-0(runs on port:9093; broker id:0) 和 broker-2(runs on port:9094; broker id:2) 正在运行。今天我错误地启动了broker-1（运行在端口：9095；broker id：1）和broker-2。停止 broker-1 并启动 broker-0 后，问题解决。现在消费者能够获取事件。

我这边肯定是人为错误，但我有 2 个 cmets：

我认为 Kafka 应该优雅地使用 broker-2(port no:9094) 而忽略 broker-1(port no:9095)
为什么 Kafka 尝试联系 4bdce773eb74:9095，而不是正确的 IP 地址 (192.168.99.100)？

谢谢。

【讨论】：