我在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();
}
我可以列出主题,也可以查看主题名称,还可以阅读主题消息。
当做
2条答案
按热度按时间kmb7vmvb1#
在我的情况下,在周末它失去了与所有节点的连接。似乎无法自动恢复连接。
重新启动服务后,一切正常。
kiayqfof2#
我猜你不是在当地经营Kafka。。。
删除此行
也可以在aws中部署schema registry,并更改此行(以及地址)
另外,如果要发送avro,请创建
GenericRecord
avro对象,而不是json字符串。否则,您的模式只是"string"