KafkaListener doesnt respect containerFactory after upgrading from spring kafka 2.7.9 to 2.8.0

[WtfJoke]

Affects Version(s): 2.8.0

After upgrading from Spring Boot 2.5.6 to 2.6.1 we noticed, that the @KafkaListener doesnt respect the containerFactory setting anymore.

In the particular case, we werent able to consume non schema registry topics anymore as the ContainerFactory with the overwritten Deserializer werent used and instead the Default Deserializer (AvroDeSerializer) was used.

I werent able to pinpoint the issue exactly, but was able to make a small demo project with which you are able to reproduce the issue.

1. Clone https://github.com/WtfJoke/springkafka28issue
2. Start app
3. curl -X POST localhost:8080/show
4. Exception thrown (Unknown Magic Byte by Avro Serializer, allthough JsonSerializer is specified in KafkaConsumerConfig#showPublishedConsumerFactory which is used by the @KafkaListener annotation

If you change spring boot version to 2.5.7 (or switch to working branch), everything works as expected and in step 4. instead of an exception, the show is logged to console.

I hope my description is clear enough, to reproduce the issue.

[artembilan]

I think the problem comes from this fix: 852c447.

So, since you have that in your config:

spring.deserializer.key.delegate.class: 'org.apache.kafka.common.serialization.StringDeserializer'
spring.deserializer.value.delegate.class: 'io.confluent.kafka.serializers.KafkaAvroDeserializer'

The ErrorHandlingDeserializer take them into account from its configure():

	public void configure(Map<String, ?> configs, boolean isKey) {
		setupDelegate(configs, isKey ? KEY_DESERIALIZER_CLASS : VALUE_DESERIALIZER_CLASS);
		Assert.state(this.delegate != null, "No delegate deserializer configured");
		this.delegate.configure(configs, isKey);
		this.isForKey = isKey;
		setupFunction(configs, isKey ? KEY_FUNCTION : VALUE_FUNCTION);
	}

We probably just need to check for this.delegate != null before calling that setupDelegate() and so on for failedDeserializationFunction.

As a workaround I suggestion you to extend that ErrorHandlingDeserializer and override its configure(Map<String, ?> configs, boolean isKey) to empty body:

    @Bean
    fun showPublishedConsumerFactory(): ConsumerFactory<String, Show> {
        val deserializer =
            object : ErrorHandlingDeserializer<Show>(JsonDeserializer(Show::class.java, false)) {
                override fun configure(configs: Map<String, *>, isKey: Boolean) {

                }
            }
        return DefaultKafkaConsumerFactory(
            consumerConfigs(), ErrorHandlingDeserializer(StringDeserializer()),
            deserializer
        )
    }

[artembilan]

I proposed the fix here: #2051

[WtfJoke]

Thank you @artembilan for your fast response + confirmation + proposed fix.
Just wanted to let you know that the proposed workaround works :)