java—60000毫秒后元数据中不存在主题

nom7f22z  于 2021-06-06  发布在  Kafka
关注(0)|答案(2)|浏览(1061)

我在msk(Kafka)中创建了一个主题。我还注册了avro模式。现在我正试图生成一个主题的消息,但是当我运行producer时,我得到了下面的错误

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Topic AVRO-AUDIT_EVENT not present in metadata after 60000 ms.

下面是我的java代码,用于生成avro消息

String topicName = "AVRO-AUDIT_EVENT";

        Properties props = new Properties();
        props.put("bootstrap.servers",
                "b-3.*****:9092,b-4.****:9092,b-5.****:9092");
        props.put("bootstrap.servers", "localhost:9092,localhost:9093");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
        props.put("schema.registry.url", "http://localhost:8081/subjects/AVRO-AUDIT-EVENT/versions/");

        JSONObject job = new JSONObject(json);
        String bodyofJson = job.getString("body");
        JSONObject bodyJsonObj = new JSONObject(bodyofJson);
        System.out.println(bodyJsonObj.get("ID"));

        Producer<String, String> producer = new KafkaProducer<>(props);
        try {

            producer.send(new ProducerRecord<String, String>(topicName, bodyJsonObj.get("ID").toString(), bodyofJson))
                    .get();

            System.out.println("Complete");
        } catch (Exception ex) {
            ex.printStackTrace(System.out);
        } finally {
            producer.close();
        }

我可以列出主题,也可以查看主题名称,还可以阅读主题消息。
当做

kmb7vmvb

kmb7vmvb1#

在我的情况下,在周末它失去了与所有节点的连接。似乎无法自动恢复连接。
重新启动服务后,一切正常。

kiayqfof

kiayqfof2#

我猜你不是在当地经营Kafka。。。
删除此行

props.put("bootstrap.servers", "localhost:9092,localhost:9093");

也可以在aws中部署schema registry,并更改此行(以及地址)

props.put("schema.registry.url", "http://localhost:8081/");

另外,如果要发送avro,请创建 GenericRecord avro对象,而不是json字符串。否则,您的模式只是 "string"

相关问题