GH-1708: Fix New Package Tangles

Resolves #1708

- add a new interface `ListenerContainerRegistry` in the `listener` package
- eliminate the `destinationtopic` sub-package
- move classes with references to the `annotation` package to `annotation`.
- add `package-info.java`

Author: garyrussell

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

@@ -78,7 +78,6 @@
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.retrytopic.RetryTopicBootstrapper;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
-import org.springframework.kafka.retrytopic.RetryTopicConfigurationProvider;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.TopicPartitionOffset;

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationProvider.java renamed to spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryTopicConfigurationProvider.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic;
+package org.springframework.kafka.annotation;
 
 import java.lang.reflect.Method;
 import java.util.Map;
@@ -25,7 +25,8 @@
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.log.LogAccessor;
-import org.springframework.kafka.annotation.RetryableTopic;
+import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
+import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
 
 
 /**

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryableTopicAnnotationProcessor.java renamed to spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic;
+package org.springframework.kafka.annotation;
 
 import java.lang.reflect.Method;
 import java.util.Arrays;
@@ -27,10 +27,11 @@
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
-import org.springframework.kafka.annotation.DltHandler;
-import org.springframework.kafka.annotation.RetryableTopic;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.listener.adapter.AdapterUtils;
+import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
+import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
+import org.springframework.kafka.retrytopic.RetryTopicInternalBeanNames;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.backoff.ExponentialBackOffPolicy;
 import org.springframework.retry.backoff.ExponentialRandomBackOffPolicy;
@@ -42,7 +43,6 @@
 import org.springframework.util.StringUtils;
 
 /**
- *
  * Processes the provided {@link RetryableTopic} annotation
  * returning an {@link RetryTopicConfiguration}.
  *

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -39,6 +39,7 @@
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.listener.ListenerContainerRegistry;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -66,8 +67,8 @@
  * @see MessageListenerContainer
  * @see KafkaListenerContainerFactory
  */
-public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifecycle, ApplicationContextAware,
-		ApplicationListener<ContextRefreshedEvent> {
+public class KafkaListenerEndpointRegistry implements ListenerContainerRegistry, DisposableBean, SmartLifecycle,
+		ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> {
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); //NOSONAR
 
@@ -96,6 +97,7 @@ public void setApplicationContext(ApplicationContext applicationContext) throws
 	 * @see KafkaListenerEndpoint#getId()
 	 * @see #getListenerContainerIds()
 	 */
+	@Override
 	public MessageListenerContainer getListenerContainer(String id) {
 		Assert.hasText(id, "Container identifier must not be empty");
 		return this.listenerContainers.get(id);
@@ -106,6 +108,7 @@ public MessageListenerContainer getListenerContainer(String id) {
 	 * @return the ids.
 	 * @see #getListenerContainer(String)
 	 */
+	@Override
 	public Set<String> getListenerContainerIds() {
 		return Collections.unmodifiableSet(this.listenerContainers.keySet());
 	}
@@ -115,6 +118,7 @@ public Set<String> getListenerContainerIds() {
 	 * @return the managed {@link MessageListenerContainer} instance(s).
 	 * @see #getAllListenerContainers()
 	 */
+	@Override
 	public Collection<MessageListenerContainer> getListenerContainers() {
 		return Collections.unmodifiableCollection(this.listenerContainers.values());
 	}
@@ -128,6 +132,7 @@ public Collection<MessageListenerContainer> getListenerContainers() {
 	 * @since 2.2.5
 	 * @see #getListenerContainers()
 	 */
+	@Override
 	public Collection<MessageListenerContainer> getAllListenerContainers() {
 		List<MessageListenerContainer> containers = new ArrayList<>();
 		containers.addAll(getListenerContainers());

spring-kafka/src/main/java/org/springframework/kafka/config/package-info.java

@@ -1,4 +1,5 @@
 /**
  * Package for kafka configuration
  */
+@org.springframework.lang.NonNullApi
 package org.springframework.kafka.config;

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

@@ -26,7 +26,6 @@
 
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationListener;
-import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.event.ListenerContainerPartitionIdleEvent;
 
 /**
@@ -36,6 +35,7 @@
  * again after partition consumption is resumed (or seek it manually by other means).
  *
  * @author Tomaz Fernandes
+ * @author Gary Russell
  * @since 2.7
  * @see SeekToCurrentErrorHandler
  */
@@ -46,14 +46,15 @@ public class KafkaConsumerBackoffManager implements ApplicationListener<Listener
 	 */
 	public static final String INTERNAL_BACKOFF_CLOCK_BEAN_NAME = "internalBackOffClock";
 
-	private final KafkaListenerEndpointRegistry registry;
+	private final ListenerContainerRegistry registry;
 
 	private final Map<TopicPartition, Context> backOffTimes;
 
 	private final Clock clock;
 
-	public KafkaConsumerBackoffManager(KafkaListenerEndpointRegistry registry,
+	public KafkaConsumerBackoffManager(ListenerContainerRegistry registry,
 									@Qualifier(INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
+
 		this.registry = registry;
 		this.clock = clock;
 		this.backOffTimes = new HashMap<>();

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import java.util.Collection;
+import java.util.Set;
+
+import org.springframework.kafka.config.KafkaListenerEndpoint;
+
+/**
+ * A registry for listener containers.
+ *
+ * @author Gary Russell
+ * @since 2.7
+ *
+ */
+public interface ListenerContainerRegistry {
+
+	/**
+	 * Return the {@link MessageListenerContainer} with the specified id or
+	 * {@code null} if no such container exists.
+	 * @param id the id of the container
+	 * @return the container or {@code null} if no container with that id exists
+	 * @see KafkaListenerEndpoint#getId()
+	 * @see #getListenerContainerIds()
+	 */
+	MessageListenerContainer getListenerContainer(String id);
+
+	/**
+	 * Return the ids of the managed {@link MessageListenerContainer} instance(s).
+	 * @return the ids.
+	 * @see #getListenerContainer(String)
+	 */
+	Set<String> getListenerContainerIds();
+
+	/**
+	 * Return the managed {@link MessageListenerContainer} instance(s).
+	 * @return the managed {@link MessageListenerContainer} instance(s).
+	 * @see #getAllListenerContainers()
+	 */
+	Collection<MessageListenerContainer> getListenerContainers();
+
+	/**
+	 * Return all {@link MessageListenerContainer} instances including those managed by
+	 * this registry and those declared as beans in the application context.
+	 * Prototype-scoped containers will be included. Lazy beans that have not yet been
+	 * created will not be initialized by a call to this method.
+	 * @return the {@link MessageListenerContainer} instance(s).
+	 * @see #getListenerContainers()
+	 */
+	Collection<MessageListenerContainer> getAllListenerContainers();
+
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/KafkaBackoffAwareMessageListenerAdapter.java

@@ -47,7 +47,9 @@
 public class KafkaBackoffAwareMessageListenerAdapter<K, V> extends AbstractDelegatingMessageListenerAdapter<AcknowledgingConsumerAwareMessageListener<K, V>> implements AcknowledgingConsumerAwareMessageListener<K, V> {
 
 	private final String listenerId;
+
 	private final String backoffTimestampHeader;
+
 	private final KafkaConsumerBackoffManager kafkaConsumerBackoffManager;
 
 	public KafkaBackoffAwareMessageListenerAdapter(AcknowledgingConsumerAwareMessageListener<K, V> delegate, KafkaConsumerBackoffManager kafkaConsumerBackoffManager, String listenerId, String backoffTimestampHeader) {

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

@@ -30,8 +30,6 @@
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.listener.KafkaBackoffException;
-import org.springframework.kafka.retrytopic.destinationtopic.DestinationTopic;
-import org.springframework.kafka.retrytopic.destinationtopic.DestinationTopicResolver;
 import org.springframework.util.Assert;
 
 /**

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic.destinationtopic;
+package org.springframework.kafka.retrytopic;
 
 import java.util.ArrayList;
 import java.util.Collection;

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/destinationtopic/DestinationTopic.java renamed to spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java

@@ -14,13 +14,12 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic.destinationtopic;
+package org.springframework.kafka.retrytopic;
 
 import java.util.Objects;
 import java.util.function.BiPredicate;
 
 import org.springframework.kafka.core.KafkaOperations;
-import org.springframework.kafka.retrytopic.DltStrategy;
 
 /**
  *

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/destinationtopic/DestinationTopicContainer.java renamed to spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicContainer.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic.destinationtopic;
+package org.springframework.kafka.retrytopic;
 
 import java.time.Clock;
 import java.time.Instant;
@@ -25,13 +25,12 @@
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.kafka.listener.ListenerExecutionFailedException;
-import org.springframework.kafka.retrytopic.RetryTopicConstants;
 
 
 /**
  *
  * Contains the destination topics and correlates them with their source via the
- * Map<String, {@link org.springframework.kafka.retrytopic.destinationtopic.DestinationTopicResolver.DestinationsHolder}> map.
+ * Map<String, {@link org.springframework.kafka.retrytopic.DestinationTopicResolver.DestinationsHolder}> map.
  *
  * Implements the {@link DestinationTopicResolver} interface.
  *

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/destinationtopic/DestinationTopicProcessor.java renamed to spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicProcessor.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic.destinationtopic;
+package org.springframework.kafka.retrytopic;
 
 import java.util.Collection;
 import java.util.HashMap;

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic.destinationtopic;
+package org.springframework.kafka.retrytopic;
 
 import java.util.Arrays;
 import java.util.List;
@@ -24,11 +24,6 @@
 
 import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.kafka.core.KafkaOperations;
-import org.springframework.kafka.retrytopic.BackOffValuesGenerator;
-import org.springframework.kafka.retrytopic.DltStrategy;
-import org.springframework.kafka.retrytopic.FixedDelayStrategy;
-import org.springframework.kafka.retrytopic.RetryTopicConstants;
-import org.springframework.kafka.retrytopic.TopicSuffixingStrategy;
 import org.springframework.retry.backoff.BackOffPolicy;
 import org.springframework.util.StringUtils;
 
@@ -195,9 +190,10 @@ private String joinWithRetrySuffix(long parameter) {
 	public static class DestinationTopicSuffixes {
 
 		private final String retryTopicSuffix;
+
 		private final String dltSuffix;
 
-		DestinationTopicSuffixes(String retryTopicSuffix, String dltSuffix) {
+		public DestinationTopicSuffixes(String retryTopicSuffix, String dltSuffix) {
 			this.retryTopicSuffix = StringUtils.hasText(retryTopicSuffix)
 					? retryTopicSuffix
 					: RetryTopicConstants.DEFAULT_RETRY_SUFFIX;

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/destinationtopic/DestinationTopicResolver.java renamed to spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicResolver.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.kafka.retrytopic.destinationtopic;
+package org.springframework.kafka.retrytopic;
 
 import java.util.Map;
 

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

@@ -25,8 +25,6 @@
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
-import org.springframework.kafka.retrytopic.destinationtopic.DefaultDestinationTopicProcessor;
-import org.springframework.kafka.retrytopic.destinationtopic.DestinationTopicContainer;
 
 /**
  *

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

@@ -18,7 +18,6 @@
 
 import java.util.List;
 
-import org.springframework.kafka.retrytopic.destinationtopic.DestinationTopic;
 import org.springframework.kafka.support.AllowDenyCollectionManager;
 
 /**