'Serialize JSON String to AVRO Apache Beam KafkaIO

I have a JSON String that needs to be serialized to AVRO format so that I can publish that to a Kafka topic that has an Avro schema with schema validation on it. Below is what I have tried so far but it gives a validation error.

@Override     
public PDone expand(PCollection<Product> input) {
    return input.apply("Write to Kafka Topic", KafkaIO.<Void, String>write().
                withBootstrapServers(kafkaUrl)
                .withTopic(kafkaTopic)
//                .withKeySerializer(StringSerializer.class)
                .withValueSerializer(StringSerializer.class)

The error I am getting is

send failed : 'This record has failed the validation on broker and hence be rejected.'
javaapache-kafkaapache-beamavroconfluent-schema-registry


Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source

Related Questions