Incorrect retry topic suffix

[Desire456]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

2.8.2

Describe the bug

Incorrect retry topic suffix (retry-5000) when attempts is equal to 2 in @RetryableTopic and fixedDelayStrategy=SINGLE_TOPIC

To Reproduce

@KafkaListener(
        id = "someId",
        topics = [GREETING_TOPIC],
        groupId = "groupId",
        autoStartup = "true"
    )
@RetryableTopic(
        backoff = Backoff(delay = 5000),
        attempts = "2",
        autoCreateTopics = "false",
        fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC
    )

Expected behavior

Retry topic suffix = -retry

[Desire456]

The problem is here:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java

Line 272 in f351a7a

return this.backOffValues.size() > 1 && this.backOffValues.stream().distinct().count() == 1;

Why this.backOffValues.size() > 1?
Is 1 attempt with fixed delay is not fixed delay according to this method? Or it's feature that I didn't realize?

[garyrussell]

Confirmed that this is a bug (when attempts = "2") but, unfortunately, we can't change the default behavior in a patch release; we could make it an opt-in, in a future release.

As a work around, see https://docs.spring.io/spring-kafka/docs/current/reference/html/#custom-naming-strategies

Here's an example:

@SpringBootApplication
public class Kgh2654Application extends RetryTopicConfigurationSupport {

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

	@Bean
	NewTopic topic() {
		return TopicBuilder.name("kgh2654").partitions(1).replicas(1).build();
	}

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

	@Override
	protected RetryTopicComponentFactory createComponentFactory() {
		return new RetryTopicComponentFactory() {

			@Override
			public RetryTopicNamesProviderFactory retryTopicNamesProviderFactory() {
				return new CustomRetryTopicNamesProviderFactory();
			}

		};
	}

	@Bean
	TaskScheduler sched() {
		return new ThreadPoolTaskScheduler();
	}

}

@Component
class Listener {

	@RetryableTopic(attempts = "2", fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)
	@KafkaListener(id = "kgh2654", topics = "kgh2654")
	void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
		System.out.println(in + " from " + topic);
		throw new RuntimeException("test");
	}

	@DltHandler
	void handle(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
		System.out.println(in + " from " + topic);
	}

}

class CustomRetryTopicNamesProviderFactory implements RetryTopicNamesProviderFactory {

	@Override
	public RetryTopicNamesProvider createRetryTopicNamesProvider(
			DestinationTopic.Properties properties) {

		if (properties.isMainEndpoint() || properties.isDltTopic()) {
			return new SuffixingRetryTopicNamesProvider(properties);
		}
		else {
			return new SuffixingRetryTopicNamesProvider(properties) {

				@Override
				public String getTopicName(String topic) {
					return topic + "-retry";
				}

			};
		}
	}

}

[Desire456]

Got it, thanks!

[iNviNho]

This is what worked for us in spring-kafka 2.8.8 👇

First we created a CustomRetryTopicNamesProviderFactory that implements RetryTopicNamesProviderFactory

import org.springframework.kafka.retrytopic.DestinationTopic;
import org.springframework.kafka.retrytopic.RetryTopicNamesProviderFactory;
import org.springframework.kafka.retrytopic.SuffixingRetryTopicNamesProviderFactory;

public class CustomRetryTopicNamesProviderFactory implements RetryTopicNamesProviderFactory {

  @Override
  public RetryTopicNamesProvider createRetryTopicNamesProvider(final DestinationTopic.Properties properties) {
    if (properties.isMainEndpoint() || properties.isDltTopic()) {
      return new SuffixingRetryTopicNamesProviderFactory.SuffixingRetryTopicNamesProvider(properties);
    } else {
      return new SuffixingRetryTopicNamesProviderFactory.SuffixingRetryTopicNamesProvider(properties) {
        @Override
        public String getTopicName(final String topic) {
          return topic + "-retry";
        }
      };
    }
  }
}

and then we created a bean for class RetryTopicNamesProviderFactory like this

package com.bitpanda.legaltenancy.infrastructure.kafka.factories;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.retrytopic.RetryTopicNamesProviderFactory;

@Configuration
public class RetryTopicComponentFactory {

  @Bean
  public RetryTopicNamesProviderFactory retryTopicNamesProviderFactory() {
    return new CustomRetryTopicNamesProviderFactory();
  }

}

Works like a charm 🥳

[garyrussell]

2.9.x and later require extending RetryTopicConfigurationSupport rather than just defining it as a bean, due to some bootstrapping problems.

[Wzy19930507]

Hi, @garyrussell

Look like the correct scenario is

• when maxAttempts is 2 and SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS --> topic-main-retry-{delay}
• when maxAttempts is 2 and SameIntervalTopicReuseStrategy.SINGLE_TOPIC --> topic-main-retry

maybe we can change code at DestinationTopicPropertiesFactory

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java

Lines 130 to 132 in 9794c58

	private boolean isSingleTopicFixedDelay() {
	return isFixedDelay() && isSingleTopicSameIntervalTopicReuseStrategy();
	}

	private boolean isSingleTopicFixedDelay() {
		return this.backOffValues.size() == 1 && isFixedDelay() && isSingleTopicSameIntervalTopicReuseStrategy();
	}

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java

Lines 222 to 225 in 9794c58

	private boolean isDelayWithReusedTopic(Long backoffValue) {
	return ((isSingleTopicFixedDelay()) ||
	(hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy()));
	}

	private boolean isDelayWithReusedTopic(Long backoffValue) {
                // `hasDuplicates(backoffValue)` contains `isFixedDelay()`
		return hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy();
	}

[Wzy19930507]

add Integration testing case for generating topicSuffix via @RetryableTopic

• diff delay maxAttempts == 2 and SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS --> topic-main-retry-{delay}
• singleFixed maxAttempts >= 2 and same backoff and SameIntervalTopicReuseStrategy.SINGLE_TOPIC --> topic-main-retry
• fixed when maxAttempts > 3 and same backoff and SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS | indexStrategy --> topic-main-retry-{index}
• muliti duplicate delay when maxAttempts > 3 and duplicates delay topic and SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS --> topic-main-retry-{delay}-{duplicatesIndex}
• single duplicate delay when maxAttempts > 3 and duplicates delay topic and SameIntervalTopicReuseStrategy.SINGLE_TOPIC --> topic-main-retry-{delay}

---

look like every backoff step in DestinationTopicPropertiesFactory#backOffValues there has two properties and five combination

delay diff as before step: add {delay}
delay same as before step: add {index}

multi: do no thing
single: delete {index}

duplication: both of diff and same: add {delay} and {index}