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

新手上路,请多包涵

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

     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");

    }
}

对于 bootstrap.server,我什至尝试了以下操作:

props.put(“bootstrap.servers”, “PLAINTEXT://192.168.xxx.xxx:9092”);

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

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

KAFKA 的版本是 2.12-2.0.0,我已将以下 jar 添加到我的 Eclipse 类路径中:

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

非常感谢您的提前帮助:)

Regadrs,DIRSHAH。

原文由 Fawad Shah 发布,翻译遵循 CC BY-SA 4.0 许可协议

阅读 1.9k
1 个回答

此错误可能表明该主题不存在,因此您可能需要仔细检查您的 topicName = “Hello-Kafka”。

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

原文由 ynux 发布,翻译遵循 CC BY-SA 4.0 许可协议

撰写回答
你尚未登录,登录后可以
  • 和开发者交流问题的细节
  • 关注并接收问题和回答的更新提醒
  • 参与内容的编辑和改进,让解决方法与时俱进
推荐问题