KAFKA:60000毫秒后无法更新元数据

Question

我是KAFKA的新手，我知道这个问题在stack overflow上被问了很多次，但没有一个解决方案对我有效，所以我在这里再次问同样的问题，试试我的运气。我已经在Centos7 VM上下载并安装了KFKA。虚拟机在我的笔记本电脑上。当我从命令行运行KAFKA生产者和消费者时，它工作得很好。下一步，我想创建一个Java生产者，但它总是超时，并出现以下异常。

    Exception in thread "main" java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
    at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:1186)
    at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:880)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803)
    at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:690)
    at com.soft.teradata.KafkaProducerExample.runProducer(KafkaProducerExample.java:40)
    at com.soft.teradata.KafkaProducerExample.main(KafkaProducerExample.java:55)
Caused by: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

生产者的 Java 代码是：

package com.soft;
import java.util.Properties;
import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class SimpleProducer {
    public static void main(String[] args) throws Exception {
        try {
            String topicName = "Hello-Kafka";
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.xxx.xxx:9092");
            props.put("acks", "all");
            props.put("retries", 1);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            Producer<String, String> producer = new KafkaProducer<String, String>(props);
            Future<RecordMetadata> f=producer.send(new ProducerRecord<String, String>(topicName, "Eclipse"));
            System.out.println("Message sent successfully");
            producer.close();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        } 
            System.out.println("Successful");

    }
} 

对于引导服务器，我甚至尝试了以下内容：

props.put（“bootstrap.servers”、“原告：//192.168.xxx.xxx:9092”）；

请注意，我正在笔记本电脑上执行Eclipse的java代码，KAFKA安装在笔记本电脑上的CENTOS7虚拟机上。192.168.xxx.xxx是CENTOS7虚拟机的IP地址。我注意到我的笔记本电脑无法访问192.168.xxx.xxx:9092（telnet192.168.xxx.xxx9092）。我将端口添加到防火墙，但仍然没有成功。

firewall-cmd --zone=public --add-port=9092/tcp --permanent
        firewall-cmd --reload
        firewall-cmd --list-ports

KAFKA的版本是2.12-2.0.0，我在Eclipse Classpath中添加了以下jar：

• kafka-clients-2.0.0.jar
• lz4-java-1.4.1.jar
• slf4j-api-1.7.25.jar
• snapy-java-1.1.7.1.jar

非常感谢你的帮助：）

迪尔沙·雷加德斯。

Answer 1

我在本地运行集群时遇到了同样的问题，我检查了主题是否是在代理中创建的：

./kafka-topics.sh --list  --bootstrap-server=localhost:9092

创建了我的代码，所以我再次检查了代码，我意识到在发送消息时主题名中有一个尾随空格，我删除了空格，它就可以工作了。

Answer 2

我也面临同样的问题。您需要通告Kafka broker的主机名/ip，以便可以从Kafka Producer pc访问。

 kafka-server-start.sh config/server.properties --override  advertised.listeners=PLAINTEXT://<accessible-hostname>:9092

Answer 3

这个错误可能表明主题不存在，所以您可能需要仔细检查您的topicName = "Hello-Kafka "。

虽然这并不是一个深刻的答案，但它似乎是一个常见的问题，另请参阅https://github.com/dpkp/kafka-python/issues/607