Kafka生产者无法连接localhost:9092问题及IPv6解决方法

问题现象 #

编写Kafka生产者示例代码时，持续出现连接错误，日志如下：

[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected
[kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Node -1 disconnected.

生产者代码 #

package io.conduktor.demos.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class ProducerDemo {

    private static final Logger log = LoggerFactory.getLogger(ProducerDemo.class.getSimpleName());

    public static void main(String[] args) {
        log.info("Hello World!");

        //create Producer Properties
        Properties properties = new Properties();

        //connect to Localhost
        properties.setProperty("bootstrap.servers", "localhost:9092");

        //connect to Conduktor playground
        //properties.setProperty("bootstrap.servers", "cluster.playground.cdkt.io:9092");
        //properties.setProperty("security.protocol", "SASL_SSL");
        //properties.setProperty("sasl.jaas.config", "org.apache.kafka...");
        //properties.setProperty("sasl.mechanism", "PLAIN");

        //set producer properties
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());

        //create the Producer
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //create a producer Record
        ProducerRecord<String, String> producerRecord =
                new ProducerRecord<>("demo_java","hello world");

        //send data
        producer.send(producerRecord);

        //tell the producer to send all data and block until done --synchronous
        producer.flush();

        //flush and close the producer
        producer.close();

    }
}

已尝试的排查操作 #

• 修改主机端口
• 修改节点ID
• 检查防火墙（未启用）

使用kafka-topics.sh查看主题的结果：

bartello@LAPTOP-K0A7F8UD:~$ kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic demo_java
Topic: demo_java        TopicId: uD7yPEAnQ6iWmQtqbJ-GCg PartitionCount: 1       ReplicationFactor: 1    Configs: segment.bytes=1073741824
        Topic: demo_java        Partition: 0    Leader: 1       Replicas: 1     Isr: 1

解决方法 #

将代码中bootstrap.servers配置项的localhost替换为[::1]，切换至IPv6连接后，问题解决，生产者可正常运行。

内容的提问来源于stack exchange，提问作者Bartello_45