AnushanTY

2018-09-19 13:11:17.560 ERROR 10535 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: null

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition trip-0 at offset 5034. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

AnushanTY

try {
            while (true) {
                ConsumerRecords<String, GenericRecord> records = null;
                try {
                    records = consumer.poll(10000);
                } catch (SerializationException e) {
                    String s = e.getMessage().split("Error deserializing key/value for partition ")[1].split(". If needed, please seek past the record to continue consumption.")[0];
                    String topics = s.split("-")[0];
                    int offset = Integer.valueOf(s.split("offset ")[1]);
                    int partition = Integer.valueOf(s.split("-")[1].split(" at")[0]);

AnushanTY

 @Bean
    public ConcurrentKafkaListenerContainerFactory<String, GenericRecord>
    kafkaListenerContainerFactory()  {



        ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory
                = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setErrorHandler(new ErrorHandler() {