Setting up the backoff aware message listener adapter should work for any message listener

[filiphr]

We had a problem here while trying to implement our programatic retry mechanism using Kafka.

This is one of the places that we had to adapt our MessageListener.

I am not sure whether the cast you require is mandatory or not, but the KafkaBackoffAwareMessageListenerAdapter does support different message listeners.

[tomazfernandes]

Hi @filiphr, this makes sense to me, we don't seem to require an AcknowledgingConsumerAwareMessageListener anywhere down the road.

As far as I can see, the consequence of having a different MessageListener for the RetryableTopics feature would be that we wouldn't be able to apply the TimingAdjustment if we don't have access to the Consumer, which doesn't seem like a deal breaker to me.

Of course, it's up to @garyrussell whether to proceed with this or not.

Thanks

[filiphr]

As far as I can see, the consequence of having a different MessageListener for the RetryableTopics feature would be that we wouldn't be able to apply the TimingAdjustment if we don't have access to the Consumer, which doesn't seem like a deal breaker to me.

Are you referring to

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java

Lines 93 to 95 in a372cc4

	maybeGetBackoffTimestamp(consumerRecord)
	.ifPresent(nextExecutionTimestamp -> this.kafkaConsumerBackoffManager
	.backOffIfNecessary(createContext(consumerRecord, nextExecutionTimestamp, consumer)));

because the KafkaBackoffAwareMessageListenerAdapter is an AcknowledgingConsumerAwareMessageListener. This change is to relax the requirement of the delegate to also be AcknowledgingConsumerAwareMessageListener. From what I could see ConcurrentMessageListenerContainer will use the newly created ConcurrentMessageListenerContainer as a message listener. Thus the right method will be invoked and you will still have access to the Consumer.

However, my knowledge about this is way more limited than yours and there might be something that I am not seeing 😄

[tomazfernandes]

Yep, I think you're right. It's the delegate that might not have access to the consumer if it's a different interface. So, no concerns from me regarding this. Thanks!

[artembilan]

We do this over there even right now:

new KafkaBackoffAwareMessageListenerAdapter<>(listener,

Which leads to this:

public KafkaBackoffAwareMessageListenerAdapter(MessageListener<K, V> adapter,

So, that is really strange to have it forced just to the AcknowledgingConsumerAwareMessageListener...

LGTM, though!

[garyrussell]

Merged as 596f7d9 and cherry-picked to 2.9.x, 2.8.x.