发送字节数组并将其反序列化为带有kafka的Avro记录

Question

我正在向kafka发送对应于Avro记录的字节数组。

生产商：

props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker-address");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put(ProducerConfig.CLIENT_ID_CONFIG, "Kafka Avro ClientOrderRequest Producer");
props.put("schema.registry.url", "schema-registry address");

KafkaProducer<String, ClientOrderRequest> producerRequest = new KafkaProducer<>(props);

        
while (true) {
    ClientOrderRequest clientOrderRequest = createClientOrderRequest();
    byte[] bytes = toByteArray(clientOrderRequest);
    final ProducerRecord<String, byte[]> producerOrderRequest = new ProducerRecord<>("client-order-request",
                "ClientOrderRequest-" + calendar.getTimeInMillis(), bytes);
    producerRequest.send(producerOrderRequest);
    producerRequest.flush();
    System.out.println("Produced 1 record.");
    Thread.sleep(2000);
}

消费者：

props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker-address");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true");
props.put("schema.registry.url", "schema-registry address");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "events-group-3");

KafkaConsumer<String, ClientOrderRequest> clientOrderRequestConsumer = new KafkaConsumer<>(props);
        clientOrderRequestConsumer.subscribe(Collections.singletonList("client-order-request"));

while (true) {
    ConsumerRecords<String, ClientOrderRequest> records = clientOrderRequestConsumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, ClientOrderRequest> record : records) {
        String key = record.key();
        ClientOrderRequest value = record.value();
            System.out.println(key);
            System.out.println(value);
        }
    }
}

生产者能够将字节数组发送到主题。但消费者无法将其反序列化。我有一个错误：

无法将B转换为com。瑞士报价。埃弗雷克斯。生成。阿夫罗。交易。客户端请求

当我生成消息时，客户端订单请求值主题已在模式注册表上注册。我知道我发送了一个字节数组，并且kafka/模式注册表期望一个Avro记录，但我希望它能够反序列化它。

如果不能使用简单的AvroKafkaSerializer，我应该实现自己的序列化程序吗？

Answer 1

什么是toByteArray？

您不需要手动序列化数据。这就是serializer类的全部要点。不清楚您为什么将此更改为之前的问题

ClientOrderRequest clientOrderRequest = createClientOrderRequest();
final ProducerRecord<String, ClientOrderRequest> producerOrderRequest = new ProducerRecord<>("client-order-request",
                "ClientOrderRequest-" + calendar.getTimeInMillis(), clientOrderRequest);
producerRequest.send(producerOrderRequest);
producerRequest.flush();

当您使用KafkaAvroSerializer.class发送字节数组时，它会发送{"type"："bytes"}的Avro对象，而不是您的自定义记录

如果确实要直接发送字节数组，可以使用ByteArraySerializer，但这会绕过架构注册表