Programmatic registration of Kafka Listeners and Retry Topics

[filiphr]

At Flowable (an Open Source Java business process management platform) we are using a programmatic approach to register Kafka Listeners using Spring Kafka. This is done using a standard (for us) json configuration, that allows users to dynamically deploy listeners during the runtime of the application. We are really happy that we were able to provide our integration using Spring Kafka, it made it really easy for us, thanks a lot for working on this library.

Until now it has been straightforward to use Spring Kafka to satisfy our limited registration use case. Recently we had requests to provide Retry / Dead Letter semantics to our integration. We had a look at the current offer from Spring Kafka and the capabilities have been amazing. Therefore, we decided to use it for this as well.

However, we hit some problems using the Java APIs for this. We managed to achieve this functionality in flowable/flowable-engine@ea5da9c and everything that is needed for Kafka is done in https://github.com/flowable/flowable-engine/tree/main/modules/flowable-event-registry-spring/src/main/java/org/flowable/eventregistry/spring/kafka. In this discussion I'd like to go over some things that we faced during the implementation, discuss potential changes in the Java API and whether you are interested in such contributions.

• RetryTopicNamesProvider depends on an impl class instead of KafkaListenerEndpoint #2294. I've already created Use KafkaListenerEndpoint instead of MethodKafkaListenerEndpoint in the RetryTopicNamesProvider #2312 to resolve that issue
• The message listener has to be AcknowledgingConsumerAwareMessageListener. I have provided Setting up the backoff aware message listener adapter should work for any message listener #2313 to fix this
• It is currently not possible to opt out of retry topics, i.e. use blocking retries with a dead letter topic only
• It is not possible to not have a dead letter handler. What we want is to send something to a dead letter, but then it is up to the user how they want to consume this. Using RetryTopicConfigurationBuilder#autoStartDltHandler does not work when the application context is refreshed because the registry starts all newly registered containers when the context has already been refreshed.
• Missing methods in the DestinationTopic.Properties e.g. there is isDltTopic and isMainTopic there, but no isAnyRetryTopic
• Creating a sleeping back off policy for the RetryTopicConfiguration had to be copied to mimic what is done in Spring Kafka (I know that this is copied from Spring Retry here)
• The way we are using DefaultDestinationTopicProcessor#processDestinationTopicProperties felt a bit strange. It feels like it is tailored a lot towards the needs of the annotation based approach. I am not sure what the best approach for this would be though.

Having said all of this. I would really like to use this opportunity to discuss whether providing an easier to use Java APIs is something that Spring Kafka would like to do. I am able to carve out some time and work on some things that would simplify our usage of the Spring Kafka Java API in regard to the retry topic configuration.

From a general point of view I think that what could be interesting is to have a way for creating a collection of KafkaListenerEndpoint(s) and their appropriate KafkaListenerContainerFactory. Something close to what RetryTopicConfigurer#processMainAndRetryListeners does, but without the registration and the link to the MethodKafkaListenerEndpoint. Perhaps there is a need to provide different customizers, that would create the KafkaListenerEndpoint, perhaps expose a RetryableKafkaListenerEndpoint that would expose some setters that are needed to copy things from the main endpoint to the new one.

Looking forward to hearing from you

[tomazfernandes]

Hi @filiphr, thanks for your interest in this.

Let me address a few points first.

• RetryTopicNamesProvider depends on an impl class instead of KafkaListenerEndpoint #2294. I've already created Use KafkaListenerEndpoint instead of MethodKafkaListenerEndpoint in the RetryTopicNamesProvider #2312 to resolve that issue
• The message listener has to be AcknowledgingConsumerAwareMessageListener. I have provided Setting up the backoff aware message listener adapter should work for any message listener #2313 to fix this

I've added a few comments on the PRs, but overall they seem ok to me.

• It is currently not possible to opt out of retry topics, i.e. use blocking retries with a dead letter topic only

Actually it is - you can set maxAttempts to 1 and configure blocking retries. Or you can skip the RetryableTopics solution altogether and configure a DeadLetterPublishingRecoverer for that case.

• It is not possible to not have a dead letter handler. What we want is to send something to a dead letter, but then it is up to the user how they want to consume this. Using RetryTopicConfigurationBuilder#autoStartDltHandler does not work when the application context is refreshed because the registry starts all newly registered containers when the context has already been refreshed.

Not sure I understand the problem here - there are quite a few ways to configure a DltHandler, both through the RetryTopicConfiguration bean and @DltHandler annotation: https://docs.spring.io/spring-kafka/docs/current/reference/html/#dlt-processing-method

• Missing methods in the DestinationTopic.Properties e.g. there is isDltTopic and isMainTopic there, but no isAnyRetryTopic

While I can see the value of such method, isn't this equivalent to !isDltTopic && !isMainTopic?

• Creating a sleeping back off policy for the RetryTopicConfiguration had to be copied to mimic what is done in Spring Kafka (I know that this is copied from Spring Retry here)

In the latest versions this has been extracted to a BackOffPolicyBuilder in the Spring Retry project, maybe you can leverage that.

• The way we are using DefaultDestinationTopicProcessor#processDestinationTopicProperties felt a bit strange. It feels like it is tailored a lot towards the needs of the annotation based approach. I am not sure what the best approach for this would be though.

Yes, this feature was implemented based on the @KafkaListener approach. Mind that I had been using the framework for maybe a month when I implemented this feature, so I wasn't even really aware instantiating containers manually was a thing 😄

Having said all of this. I would really like to use this opportunity to discuss whether providing an easier to use Java APIs is something that Spring Kafka would like to do. I am able to carve out some time and work on some things that would simplify our usage of the Spring Kafka Java API in regard to the retry topic configuration.

As far as I'm aware, the team is looking into making some redesign work for this feature for the upcoming 3.0 version. As a first time framework contributor, while the feature seems to be working quite well and thankfully probably helping out lots of users all around, there were some design decisions I made that are not in sync with Spring standards or simply not great overall, as you've been noticing.

So while the simple fixes you're proposing in the PRs seems harmless enough for me, I'm not sure the team will want to invest more of their (and yours) time into this before such redesign is in place - which should eventually fix many of the issues you've been facing.

Of course, that's up to @garyrussell and @artembilan - I should be available for input should this proceed.

Thanks.

[garyrussell]

We are very much interested in your valuable input to make this feature easier to use/configure.

One comment on your points:

... because the registry starts all newly registered containers when the context has already been refreshed.

2.8.7 added a new boolean to the KLER to not auto start containers after the context refresh: #2301

[tomazfernandes]

Glad to hear that @garyrussell, I’m sure @filiphr will be able to make great contributions to the feature.

And thanks @filiphr for looking into this! Let me know if there’s anything I can help with.

[artembilan]

Here is mine 👍 to look into some redesign.
As you mentioned before it is not clear how to combine Spring Integration dynamic channel adapter registration and retry topic configuration.
So, if we fix that gap over here in Spring Kafka, everyone would benefit 😄 .
Unfortunately I'm still learning how this pattern is implemented over here, so I cannot give any advice how to proceed: @tomazfernandes is the best person to consult.

Thank you!

[tomazfernandes]

One quick note - @filiphr, I'm not sure what version you're looking at, perhaps 2.8.x - in 2.9 / 3.0 we've made a couple of improvements worth noting.

The first is an improved error handling where the container doesn't poll again for records after a successful recovery - that's an opt-in feature that should improve overall performance, and specifically for RetryableTopics since we always recover from errors there by sending the record to a different topic.

The other is a revamped bootstrap for the RetryableTopic configuration, mainly in the RetryTopicConfigurationSupport. I'm not sure if in your strategy it makes sense to use this class, but it should offer some insight on how the components are related in case you haven't seen it already.

Thanks