I have a consumer app for consuming a Kafka topic.
When the producer publishes the record of class type Reservation through Avro, it assigns the schema name as my_reservations on the schema registry.
In my consumer app, I set the config as
default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
and the processor as
@Bean
public java.util.function.Consumer<KStream<String, Reservation>> process() {
return input.foreach((key, value) -> {
System.out.println(" Value: " + value);
});
}
But I'm getting below error from the consumer because the Reservation classpath is reservation.Reservation.
Caused by: org.apache.kafka.common.errors.SerializationException: Could not find class reservation.my_reservations specified in writer's schema whilst finding reader's schema for a SpecificRecord.
How can I fix it? I'm using the functional way to create the consumer as mentioned in the official doc.