Spring Cloud Stream kafka-streams application shuts down on deserialization exception

I am using

spring boot: 2.3.5.RELEASE spring cloud: Hoxton.SR8

I am trying spring cloud stream kafka-streams application. Everything was running fine till got a deserialization exception. Application shuts down everytime.

I want to skip bad records and move ahead in Kafka topic. But I am not able to achieve this. Configuration:

spring:
  application:
    name: statsprocessor.${ENV}.${INSTANCE_ID}
  cloud:
    stream:
      instance-index: ${INSTANCE_INDEX}
      instance-count: ${INSTANCE_COUNT}
      bindings:
        statsInput:
          destination: ${STORE_INPUT_TOPIC}
          group: statsprocessor.${ENV}
          consumer:
            concurrency: ${CONCURRENCY}
            partitioned: true
            useNativeDecoding: true
      kafka:
        streams:
          bindings:
            statsInput:
              consumer:
                keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
                valueSerde: per.shades.framework.kafka.serdes.CotsEventSerde
                startOffset: earliest
                applicationId: statsprocessor.${ENV}
                autoCommitOnError: false
                dlqName: ${STORE_INPUT_DLQ}
                useNativeDecoding: true
                configuration:
                  client.id: statsprocessor.${ENV}.${INSTANCE_ID}
          binder:
            auto-add-partitions: true
            auto-create-topics: true
            deserializationExceptionHandler: logAndContinue
            brokers:
              - ${KAFKA_URI}
            configuration:
              num.stream.threads: ${CONCURRENCY}
              buffered.records.per.partition: 500
              cache.max.bytes.buffering: 10485760
              commit.interval.ms: 500
              state.dir: ${KAFKA_STATE_DIR}
              replication.factor: ${DEFAULT_KAFKA_TOPIC_REPLICATION_FACTOR}
              reconnect.backoff.ms: 15000
              retry.backoff.ms: 10000
              producer.linger.ms: 100
              producer.acks: all
              producer.retries: 3
              producer.batch.size: 16384
              consumer.max.poll.records: 100
              consumer.session.timeout.ms: 60000

The error I get is

    Exception in thread "statsprocessor.local.1-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: 
Exception caught in process. taskId=0_4, processor=KSTREAM-SOURCE-0000000000, topic=cots-event-store, partition=4, offset=0, stacktrace=org.apache.kafka.streams.errors.StreamsException:
 ClassCastException invoking Processor. Do the Processor's input types match the deserialized types? Check the Serde setup and change the default Serdes in StreamConfig or provide correct Serdes via method parameters. 
Make sure the Processor can accept the deserialized input of type key: unknown because key is null, and value: per.shades.model.events.CotsEvent.

Note that although incorrect Serdes are a common cause of error, the cast exception might have another cause (in user code, for example). For example, if a processor wires in a store, but casts the generics incorrectly, a class cast exception could be raised during processing, but the cause would not be wrong Serdes.
    ....
    ....
    Caused by: java.lang.ClassCastException: class per.shades.model.events.CotsEvent cannot be cast to class per.shades.model.stats.StatsMetadata (per.shades.model.events.CotsEvent and per.shades.model.stats.StatsMetadata are in unnamed module of loader 'app')

Now I am using this setting deserializationExceptionHandler: logAndContinue. Still it has no effect. As per documentation it should simply log error and continue processing. i.e it should skip bad record. But that doesnt happen. This error is seen.

All stream threads have died. The instance will be in error state and should be closed.

I also used

@Bean
public StreamsBuilderFactoryBeanCustomizer streamsBuilderFactoryBeanCustomizer()
{
    return streamsBuilderFactoryBean ->
    {
        streamsBuilderFactoryBean.getStreamsConfiguration()                    .put(org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
                        ContinueOnErrorHandler.class);
    };
}

And handler class is

public class ContinueOnErrorHandler implements DeserializationExceptionHandler
{

    @Override
    public DeserializationHandlerResponse handle(ProcessorContext processorContext, ConsumerRecord<byte[], byte[]> consumerRecord, Exception e)
    {
        System.out.println(">>>>>>> We are here");
        return DeserializationHandlerResponse.CONTINUE;
    }

    @Override
    public void configure(Map<String, ?> map)
    {

    }
}

But this also is not working. It is not getting called.

I dont want to delete my Kafka topic to get rid of bad record. Really struggling to solve simple deserialization error. Please help !!

Edit: bindings code:

@Configuration
public interface StatsStreamBindings
{
    String statsInput = "statsInput";

    @Input(statsInput)
    KStream<String, StatsMetadata> statsInput();

}

processor signature

public void aggregateStats(KStream<String, StatsMetadata> inputStream)