Custom (De)Serializer with Dead Letter Queue does not work

[cj19]

Hi!

I am using this example project to investigate the possibilities of the DLQ fault strategy.

My configuration looks like this:

quarkus.http.port=8070
kafka.bootstrap.servers=localhost:9092

# single upstream producer
mp.messaging.outgoing.events.connector=smallrye-kafka
mp.messaging.outgoing.events.topic=ctl-inbound-generic
mp.messaging.outgoing.events.value.serializer=org.apache.kafka.common.serialization.StringSerializer

# single upstream consumer
mp.messaging.incoming.generic-events.connector=smallrye-kafka
mp.messaging.incoming.generic-events.topic=ctl-inbound-generic
mp.messaging.incoming.generic-events.auto.offset.reset=earliest
**mp.messaging.incoming.generic-events.value.deserializer=com.pdelert.CustomDeserializer**
mp.messaging.incoming.generic-events.failure-strategy=dead-letter-queue
mp.messaging.incoming.generic-events.dead-letter-queue.topic=ctl-inbound-generic-dlq

# single upstream dlq consumer
mp.messaging.incoming.ctl-inbound-generic-dlq.connector=smallrye-kafka
mp.messaging.incoming.ctl-inbound-generic-dlq.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

You can see here that i change the deserializer for the generic events channel. The CustomDeserializer looks like the following, it's simple class only created for testing the configuration.

import org.apache.kafka.common.serialization.Deserializer;

public class CustomDeserializer implements Deserializer {

@Override
public Object deserialize(String s, byte[] bytes) {
    return s;
}

}

With this I get the following error:

org.apache.kafka.common.config.ConfigException: Invalid value com.pdelert.CustomSerializer for configuration value.serializer: Class com.pdelert.CustomSerializer could not be found.

Which is odd....I thought the problem could be the name and it does not implement the Serializable interface.

public class NewSerializer implements Serializable, Deserializer<Object> {
    @Override
    public Object deserialize(String s, byte[] bytes) {
        return s;
    }
}

The error is the following now:

org.apache.kafka.common.KafkaException: class com.pdelert.NewSerializer is not an instance of org.apache.kafka.common.serialization.Serializer

What am I doing wrong?