GH-2282: Fix Multi RetryableTopic Same Topic

Resolves #2282

* Fix docs.

* Copyrights; author tags.

* Remove unnecessary method override in KLEAdapter.

* Fix new test after rebase.

* Fix test after rebase. Move `getEndpointId()` call to within `ListenerScope`.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -775,6 +775,37 @@ public class CustomRetryTopicNamesProviderFactory implements RetryTopicNamesProv
 ----
 ====
 
+[[multi-retry]]
+==== Multiple Listeners, Same Topic(s)
+
+Starting with version 3.0, it is now possible to configure multiple listeners on the same topic(s).
+In order to do this, you must use custom topic naming to isolate the retry topics from each other.
+This is best shown with an example:
+
+====
+[source, java]
+----
+@RetryableTopic(...
+        retryTopicSuffix = "-listener1", dltTopicSuffix = "-listener1-dlt",
+        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
+@KafkaListener(id = "listener1", groupId = "group1", topics = TWO_LISTENERS_TOPIC, ...)
+void listen1(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
+    ...
+}
+
+@RetryableTopic(...
+        retryTopicSuffix = "-listener2", dltTopicSuffix = "-listener2-dlt",
+        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
+@KafkaListener(id = "listener2", groupId = "group2", topics = TWO_LISTENERS_TOPIC, ...)
+void listen2(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
+    ...
+}
+----
+====
+
+The `topicSuffixingStrategy` is optional.
+The framework will configure and use a separate set of retry topics for each listener.
+
 ==== Dlt Strategies
 
 The framework provides a few strategies for working with DLTs. You can provide a method for DLT processing, use the default logging method, or have no DLT at all. Also you can choose what happens if DLT processing fails.

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -40,6 +40,10 @@ You can now set a different `concurrency` for the retry containers; by default,
 
 See <<retry-config>> for more information.
 
+You can now configure multiple `@RetryableTopic` listeners on the same topic in the same application context.
+Previously, this was not possible.
+See <<multi-retry>> for more information.
+
 [[x30-lc-changes]]
 ==== Listener Container Changes
 

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -468,6 +468,7 @@ private void processMultiMethodListeners(Collection<KafkaListener> classLevelLis
 					new MultiMethodKafkaListenerEndpoint<>(checkedMethods, defaultMethod, bean);
 			String beanRef = classLevelListener.beanRef();
 			this.listenerScope.addListener(beanRef, bean);
+			endpoint.setId(getEndpointId(classLevelListener));
 			processListener(endpoint, classLevelListener, bean, beanName, resolveTopics(classLevelListener),
 					resolveTopicPartitions(classLevelListener));
 			this.listenerScope.removeListener(beanRef);
@@ -478,6 +479,7 @@ protected void processKafkaListener(KafkaListener kafkaListener, Method method,
 		Method methodToUse = checkProxy(method, bean);
 		MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
 		endpoint.setMethod(methodToUse);
+		endpoint.setId(getEndpointId(kafkaListener));
 
 		String beanRef = kafkaListener.beanRef();
 		this.listenerScope.addListener(beanRef, bean);
@@ -627,7 +629,6 @@ private void processKafkaListenerAnnotation(MethodKafkaListenerEndpoint<?, ?> en
 
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
-		endpoint.setId(getEndpointId(kafkaListener));
 		endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));
 		endpoint.setTopicPartitions(tps);
 		endpoint.setTopics(topics);

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -356,7 +356,8 @@ public void afterPropertiesSet() {
 	public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		C instance = createContainerInstance(endpoint);
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
+				.acceptIfNotNull(endpoint.getId(), instance::setBeanName)
+				.acceptIfNotNull(endpoint.getMainListenerId(), instance::setMainListenerId);
 		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
 			configureEndpoint((AbstractKafkaListenerEndpoint<K, V>) endpoint);
 		}

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -120,6 +120,9 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private String correlationHeaderName;
 
+	@Nullable
+	private String mainListenerId;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -154,6 +157,16 @@ public void setId(String id) {
 		this.id = id;
 	}
 
+	public void setMainListenerId(String id) {
+		this.mainListenerId = id;
+	}
+
+	@Override
+	@Nullable
+	public String getMainListenerId() {
+		return this.mainListenerId;
+	}
+
 	@Nullable
 	@Override
 	public String getId() {

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java

@@ -155,4 +155,14 @@ default byte[] getListenerInfo() {
 		return null;
 	}
 
+	/**
+	 * Return the main listener id if this container is for a retry topic.
+	 * @return the main listener id or null.
+	 * @since 3.0
+	 */
+	@Nullable
+	default String getMainListenerId() {
+		return null;
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -119,6 +119,9 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	private volatile boolean stoppedNormally = true;
 
+	@Nullable
+	private String mainListenerId;
+
 	/**
 	 * Construct an instance with the provided factory and properties.
 	 * @param consumerFactory the factory.
@@ -376,6 +379,21 @@ public String getListenerId() {
 		return this.beanName; // the container factory sets the bean name to the id attribute
 	}
 
+	/**
+	 * Set the main listener id, if this container is for a retry topic.
+	 * @param id the id.
+	 * @since 3.0.
+	 */
+	public void setMainListenerId(String id) {
+		this.mainListenerId = id;
+	}
+
+	@Override
+	@Nullable
+	public String getMainListenerId() {
+		return this.mainListenerId;
+	}
+
 	@Nullable
 	@Override
 	public byte[] getListenerInfo() {

spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java

@@ -196,6 +196,17 @@ default String getListenerId() {
 		throw new UnsupportedOperationException("This container does not support retrieving the listener id");
 	}
 
+	/**
+	 * The 'id' attribute of the main {@code @KafkaListener} container, if this container
+	 * is for a retry topic; null otherwise.
+	 * @return the id.
+	 * @since 3.0
+	 */
+	@Nullable
+	default String getMainListenerId() {
+		throw new UnsupportedOperationException("This container does not support retrieving the main listener id");
+	}
+
 	/**
 	 * Get arbitrary static information that will be added to the
 	 * {@link KafkaHeaders#LISTENER_INFO} header of all records.

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

@@ -24,6 +24,7 @@
 import java.util.Set;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
+import java.util.function.Function;
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -158,10 +159,10 @@ public void alwaysLogListenerException() {
 	}
 
 	@SuppressWarnings("unchecked")
-	public DeadLetterPublishingRecoverer create() {
+	public DeadLetterPublishingRecoverer create(String mainListenerId) {
+		Assert.notNull(mainListenerId, "'listenerId' cannot be null");
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(// NOSONAR anon. class size
-				this::resolveTemplate,
-				false, (this::resolveDestination)) {
+				templateResolver(mainListenerId), false, destinationResolver(mainListenerId)) {
 
 			@Override
 			protected DeadLetterPublishingRecoverer.HeaderNames getHeaderNames() {
@@ -185,7 +186,8 @@ protected DeadLetterPublishingRecoverer.HeaderNames getHeaderNames() {
 			}
 		};
 
-		recoverer.setHeadersFunction((consumerRecord, e) -> addHeaders(consumerRecord, e, getAttempts(consumerRecord)));
+		recoverer.setHeadersFunction(
+				(consumerRecord, e) -> addHeaders(mainListenerId, consumerRecord, e, getAttempts(consumerRecord)));
 		if (this.headersFunction != null) {
 			recoverer.addHeadersFunction(this.headersFunction);
 		}
@@ -199,33 +201,35 @@ protected DeadLetterPublishingRecoverer.HeaderNames getHeaderNames() {
 		return recoverer;
 	}
 
-	private KafkaOperations<?, ?> resolveTemplate(ProducerRecord<?, ?> outRecord) {
-		return this.destinationTopicResolver
-						.getDestinationTopicByName(outRecord.topic())
+	private Function<ProducerRecord<?, ?>, KafkaOperations<?, ?>> templateResolver(String mainListenerId) {
+		return outRecord -> this.destinationTopicResolver
+						.getDestinationTopicByName(mainListenerId, outRecord.topic())
 						.getKafkaOperations();
 	}
 
 	public void setDeadLetterPublishingRecovererCustomizer(Consumer<DeadLetterPublishingRecoverer> customizer) {
 		this.recovererCustomizer = customizer;
 	}
 
-	private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e) {
-		if (SeekUtils.isBackoffException(e)) {
-			throw (NestedRuntimeException) e; // Necessary to not commit the offset and seek to current again
-		}
+	private BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver(String mainListenerId) {
+		return (cr, ex) -> {
+			if (SeekUtils.isBackoffException(ex)) {
+				throw (NestedRuntimeException) ex; // Necessary to not commit the offset and seek to current again
+			}
 
-		DestinationTopic nextDestination = this.destinationTopicResolver.resolveDestinationTopic(
-				cr.topic(), getAttempts(cr), e, getOriginalTimestampHeaderLong(cr));
+			DestinationTopic nextDestination = this.destinationTopicResolver.resolveDestinationTopic(mainListenerId,
+					cr.topic(), getAttempts(cr), ex, getOriginalTimestampHeaderLong(cr));
 
-		LOGGER.debug(() -> "Resolved topic: " + (nextDestination.isNoOpsTopic()
-				? "none"
-				: nextDestination.getDestinationName()));
+			LOGGER.debug(() -> "Resolved topic: " + (nextDestination.isNoOpsTopic()
+					? "none"
+					: nextDestination.getDestinationName()));
 
-		maybeLogListenerException(e, cr, nextDestination);
+			maybeLogListenerException(ex, cr, nextDestination);
 
-		return nextDestination.isNoOpsTopic()
-					? null
-					: resolveTopicPartition(cr, nextDestination);
+			return nextDestination.isNoOpsTopic()
+						? null
+						: resolveTopicPartition(cr, nextDestination);
+		};
 	}
 
 	private void maybeLogListenerException(Exception e, ConsumerRecord<?, ?> cr, DestinationTopic nextDestination) {
@@ -299,25 +303,27 @@ else if (value.length == Integer.BYTES) {
 		return 1;
 	}
 
-	private Headers addHeaders(ConsumerRecord<?, ?> consumerRecord, Exception e, int attempts) {
+	private Headers addHeaders(String mainListenerId, ConsumerRecord<?, ?> consumerRecord, Exception e, int attempts) {
 		Headers headers = new RecordHeaders();
 		byte[] originalTimestampHeader = getOriginalTimestampHeaderBytes(consumerRecord);
 		headers.add(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP, originalTimestampHeader);
 		headers.add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS,
 				ByteBuffer.wrap(new byte[Integer.BYTES]).putInt(attempts + 1).array());
 		headers.add(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP,
-				BigInteger.valueOf(getNextExecutionTimestamp(consumerRecord, e, originalTimestampHeader))
+				BigInteger
+						.valueOf(getNextExecutionTimestamp(mainListenerId, consumerRecord, e, originalTimestampHeader))
 						.toByteArray());
 		return headers;
 	}
 
-	private long getNextExecutionTimestamp(ConsumerRecord<?, ?> consumerRecord, Exception e,
+	private long getNextExecutionTimestamp(String mainListenerId, ConsumerRecord<?, ?> consumerRecord, Exception e,
 			byte[] originalTimestampHeader) {
 
 		long originalTimestamp = new BigInteger(originalTimestampHeader).longValue();
 		long failureTimestamp = getFailureTimestamp(e);
-		long nextExecutionTimestamp =  failureTimestamp + this.destinationTopicResolver
-				.resolveDestinationTopic(consumerRecord.topic(), getAttempts(consumerRecord), e, originalTimestamp)
+		long nextExecutionTimestamp = failureTimestamp + this.destinationTopicResolver
+				.resolveDestinationTopic(mainListenerId, consumerRecord.topic(), getAttempts(consumerRecord), e,
+						originalTimestamp)
 				.getDestinationDelay();
 		LOGGER.debug(() -> String.format("FailureTimestamp: %s, Original timestamp: %s, nextExecutionTimestamp: %s",
 				failureTimestamp, originalTimestamp, nextExecutionTimestamp));

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,6 +27,7 @@
  * Default implementation of the {@link DestinationTopicProcessor} interface.
  *
  * @author Tomaz Fernandes
+ * @author Gary Russell
  * @since 2.7
  *
  */
@@ -60,7 +61,8 @@ public void processRegisteredDestinations(Consumer<Collection<String>> topicsCal
 		context
 				.destinationsByTopicMap
 				.values()
-				.forEach(topicDestinations -> this.destinationTopicResolver.addDestinationTopics(topicDestinations));
+				.forEach(topicDestinations -> this.destinationTopicResolver.addDestinationTopics(
+						context.listenerId, topicDestinations));
 		topicsCallback.accept(getAllTopicsNamesForThis(context));
 	}