I have supplied a bean for retry topic config:
@Bean
public RetryTopicConfiguration kafkaRetryTopicConfig(...) {
return RetryTopicConfigurationBuilder
.newInstance()
.fixedBackOff(...)
.maxAttempts(..)
.useSingleTopicForFixedDelays()
.doNotRetryOnDltFailure()
.listenerFactory(factory)
.create(template);
}
I have also defined a bean for retry topic names provider factory:
// Need this because it is not just retry/DLT topic suffix but the entire name is configurable and specified in the configuration.
@Bean
public RetryTopicNamesProviderFactory retryTopicNamingProviderFactory() {
return new RetryTopicNamesProviderFactory {
@Override
public RetryTopicNamesProvider createRetryTopicNamesProvider(DestinationTopic.Properties properties) {
return new SuffixingRetryTopicNamesProvider(properties) {
@Override
public String getTopicName(String topic) {
// NOTE: The retry and dead letter topic names come from config.
// Like to create different names for retry and dead letter topics.
// TODO: But how to distinguish which name I should return here!
return *******;
}
};
}
}
}
Two-part question:
a. Spring does not seem to correlate the beans. It is processing kafkaRetryTopicConfig
bean but is not using the retryTopicNamingProviderFactory
and custom topic names. It takes the default naming (.retry
and .deadLetter
). Is there any bean or component that I should declare so that the custom naming provider is picked up?
b. As indicated (TODO
) in the code above, how do I distinguish whether createRetryTopicNamesProvider
is called for retry or dead letter topic naming? I see that RetryTopicNamesProviderFactory
handles naming for both retry and dead letter topics.
Help much appreciated.
For spring boot 3, RetryTopicComponentFactory
could be overwritten:
@Bean
public RetryTopicComponentFactory customRetryTopicComponentFactory() {
return new RetryTopicComponentFactory() {
@Override
public RetryTopicNamesProviderFactory retryTopicNamesProviderFactory() {
return new CustomRetryTopicNamesProviderFactory();
}
};
}