我已经在MSK(Kafka)中创建了一个主题。而且我已经注册了avro模式。现在,我正在尝试生成有关该主题的消息,但是当我运行生产者时,出现以下错误
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();
}
我可以列出主题,也可以看到主题名称,我也可以阅读主题消息。但是当我出现问题时,会出现此错误。
问候
我假设您不在本地运行Kafka ...
删除此行
props.put("bootstrap.servers", "localhost:9092,localhost:9093");
也许也可以在AWS中部署架构注册表,并更改此行(以及地址)
props.put("schema.registry.url", "http://localhost:8081/");
[此外,如果要发送Avro,请创建GenericRecord
Avro对象,而不是JSON字符串。否则,您的架构就是"string"