Exception when combining kafka blocking/non-blocking retries

[gemi2010]

Hello,

Currently, i'm configuring both blocking/non-blocking so according to the documentation i created the following Configuration

@Slf4j
@Configuration
@EnableKafka
public class KafkaRetryConfig extends RetryTopicConfigurationSupport {
	
	@Value("${kafka.retry.maxRetryAttempts}")
	private int maxRetryAtempts;
	
	@Value("${kafka.retry.retryIntervalMilliseconds}")
	private int retryIntervalMillis;
	
	@Value("${kafka.retry.timeout}")
	private int retryTimeout;
	
	@Value("${kafka.topic.name}")
	private String topicName;
	
	private static final String RETRY_TOPIC_SUFFIX = ".retry";
    private static final String DLT_TOPIC_SUFFIX = ".DLT";
	
	
	@Bean
    public RetryTopicConfiguration attRetryableTopic(@Qualifier("retryableKafkaTemplate")KafkaTemplate<String, CustomMessage> retryableKafkaTemplate) {
		
        return RetryTopicConfigurationBuilder
                .newInstance()
                .fixedBackOff(retryIntervalMillis)
                .maxAttempts(maxRetryAtempts)
                .timeoutAfter(retryTimeout)
                .includeTopic(topicName)
                .listenerFactory("retryKafkaListenerContainerFactory")
                .sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
                .retryTopicSuffix(RETRY_TOPIC_SUFFIX)
                .dltSuffix(DLT_TOPIC_SUFFIX)
                .doNotAutoCreateRetryTopics()
                .create(retryableKafkaTemplate);
    }
	@Override
	protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetries) {
	    blockingRetries
	            .retryOn(WebClientRequestException.class)
	            .backOff(new FixedBackOff(retryIntervalMillis, maxRetryAtempts));
	}

	@Override
	protected void manageNonBlockingFatalExceptions(List<Class<? extends Throwable>> nonBlockingFatalExceptions) {
	    nonBlockingFatalExceptions.addAll(Arrays.asList(MessageHandlingException.class, SerializationException.class, NullPointerException.class,
                IllegalArgumentException.class, DataIntegrityViolationException.class, DeserializationException.class));
	}
	
	@Override
    protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
        // Use the new 2.9 mechanism to avoid re-fetching the same records after a pause
        customizersConfigurer.customizeErrorHandler(eh -> {
            eh.setSeekAfterError(false);
        });
    }
	
	@Override
    protected Consumer<DeadLetterPublishingRecovererFactory> configureDeadLetterPublishingContainerFactory() {
        return dlprf -> dlprf.setRetainAllRetryHeaderValues(false);
    }
}

for some reason i got the following error:
Unsatisfied dependency expressed through method 'retryTopicConfigurer' parameter 0: Error creating bean with name 'org.springframework.kafka.config.internalKafkaConsumerBackOffManager' , Failed to instantiate [org.springframework.kafka.listener.KafkaConsumerBackoffManager]: Factory method 'kafkaConsumerBackoffManager' threw exception with message: Either a RetryTopicSchedulerWrapper or TaskScheduler bean is required

Could some give me insight about what's missing?

[garyrussell]

Either a RetryTopicSchedulerWrapper or TaskScheduler bean is required

Simply add a TaskScheduler bean - it is used to schedule partition resumption after the delay.

See the documentation: https://docs.spring.io/spring-kafka/docs/current/reference/html/#programmatic-construction

[gemi2010]

Got it, but now i have another issue is that the RetryTopicConfiguration is always taking precedence so the blocking retries is always ignored and the non-blocking retries is in use, is there anything missing in the above config ?

[garyrussell]

It looks ok; hard to say what might be wrong looking at static configuration; this works as expected...

@SpringBootApplication
@EnableScheduling
public class Kgh2658Application extends RetryTopicConfigurationSupport {

	public static void main(String[] args) {
		SpringApplication.run(Kgh2658Application.class, args);
	}

	@Bean
	ApplicationRunner runner(KafkaTemplate<String, String> template) {
		return args -> {
			template.send("kgh2658", "foo");
		};
	}

	@Override
	protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetries) {
		blockingRetries
				.retryOn(IllegalStateException.class)
				.backOff(new FixedBackOff(5000L, 2));
	}

	@Override
	protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
		// Use the new 2.9 mechanism to avoid re-fetching the same records after a pause
		customizersConfigurer.customizeErrorHandler(eh -> {
			eh.setSeekAfterError(false);
		});

	}

}

@Component
class Listener {

	@KafkaListener(id = "kgh2658", topics = "kgh2658")
	@RetryableTopic
	void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
		System.out.println(in + " from " + topic);
		if ("kgh2658".equals(topic)) {
			throw new IllegalStateException("test");
		}
		else {
			throw new IllegalArgumentException();
		}
	}

}

foo from kgh2658
foo from kgh2658
foo from kgh2658
foo from kgh2658-retry-0
foo from kgh2658-retry-1
Received message in dlt listener: kgh2658-dlt-0@2

If you can't figure it out, provide a small, complete, example similar to mine that exhibits the behavior, we can take a look at what's wrong.

https://stackoverflow.com/help/minimal-reproducible-example

[gemi2010]

Thanks for the example you provided, so my question is this example using blocking or non-blocking retry ?
because i want to combine both so if you can provide an example that has both blocking and non-blocking configured together, that would help me a lot.

[garyrussell]

Both; when it throws an IllegalStateException we get blocking retries (3 delivery attempts and the configured 2 retries - there was an extra foo from kgh2658 in my original reply); when the retries are exhausted, it goes non-blocking; hence the next delivery is from the first retry topic.

If you want only blocking retries for certain exceptions, and don't want to switch, use Long.MAX_VALUE in maxAttempts, but that kind of defeats the whole purpose.

[gemi2010]

Thanks a lot for your help, that aligned with what i was looking for.