Kafka中的Schemanotfoundexception

此异常是由于Kafka消息包含的数据不符合正常的schema而引起的。解决此问题的方法是确保正确加载和注册相关的schema，建议使用Confluent Schema Registry来管理schema。

以下是Java代码示例：

// 创建并配置Schema Registry客户端 Properties props = new Properties(); props.put("schema.registry.url", "http://localhost:8081"); SchemaRegistryClient client = new CachedSchemaRegistryClient(props);

// 获取指定主题对应的schema String topic = "my_topic"; int version = 1; Schema schema = client.getByVersion(topic + "-value", version);

// 使用schema创建并序列化消息 GenericRecord record = new GenericData.Record(schema); record.put("field1", "value1"); record.put("field2", 123); ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) { new KafkaAvroSerializer(client).serialize(topic, record, dataOutputStream); } byte[] bytes = outputStream.toByteArray();

// 发送消息到Kafka ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, bytes); producer.send(record); producer.flush();