Combining blocking and non-blocking retries

[Casper-dss]

Hi

We want to user either blocking or non-blocking retries, depending on the exception, for the same topic. Ex. for database exceptions we would like to use blocking retries, and for ex. weblient exceptions, such as WebClientResponseException.InternalServerError, we would like to use non-blocking retries.

[artembilan]

Well, I would go just Spring Retry (@Retryable on your @KafkaListener method), instead of trying to come up with some working solution via framework config.
This way, you would have a blocking retry on your method for those specific exceptions, if they are not retryable by the @Retryable , then it they are going to be bubbled up to the listener container where non-blocking retry might pick them up.

[artembilan]

Correct. The ListenerContainerFactoryConfigurer from the retrytopic package does this in its RetryTopicListenerContainerFactoryDecorator.createListenerContainer():

CommonErrorHandler errorHandler = createErrorHandler(
					ListenerContainerFactoryConfigurer.this.deadLetterPublishingRecovererFactory
							.create(mainListenerId));
...
			listenerContainer
					.setCommonErrorHandler(errorHandler);

So, whatever is there for retry topic is overriding what was provided for the container before.
From the framework standpoint the blocking and non-blocking retries cannot be combined.

You may consider to override its:

protected CommonErrorHandler createErrorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {

for some other "mixed" CommonErrorHandler, but for me it feels much easier and more natural to go with a @retryable on the @KafkaListener method.

[Casper-dss]

The reason why we are not using @retryable is, we have created a Kafka starter library with as much configuration as possible, so our developers doesn't have to do much, when creating a new service with @KafkaListeners.

Can you point me in a direction for overriding it?

[artembilan]

The @Retryable has this option:

	/**
	 * Retry interceptor bean name to be applied for retryable method. Is mutually
	 * exclusive with other attributes.
	 * @return the retry interceptor bean name
	 */
	String interceptor() default "";

So, you can configure a RetryOperationsInterceptor bean in your library and have it injected on your meta-annotation for the @KafkaListener to be used in target projects.

[Casper-dss]

Could this in the future be possible, in a more smooth way. I think it is a bit confusing, blocking and non-blocking is configured so differently, and the possibility to use both, on the same topic, depending on the exception, is almost impossible.

[artembilan]

It looks like you can achieve you goal with a CommonDelegatingErrorHandler: https://docs.spring.io/spring-kafka/reference/kafka/annotation-error-handling.html#cond-eh.
And its JavaDocs:

 * An error handler that delegates to different error handlers, depending on the exception
 * type. The delegates must have compatible properties ({@link #isAckAfterHandle()} etc.
 * {@link #deliveryAttemptHeader()} is not supported - always returns false.

Then you override the mentioned ListenerContainerFactoryConfigurer.createErrorHandler() to produce this CommonDelegatingErrorHandler instance instead.
Then you extend a RetryTopicComponentFactory to override its listenerContainerFactoryConfigurer() to return your custom ListenerContainerFactoryConfigurer.
And expose it as a bean.
That is what available from the RetryTopicConfigurationSupport via @EnableRetryTopic.

At the same time I see this doc: https://docs.spring.io/spring-kafka/reference/retrytopic/retry-topic-combine-blocking.html.

Can you tell us why that solution doesn't work for you?