GH-1158: Fix TopicPartitionInitialOff Deprecation

Fixes #1158

Revert "GH-1158: Deprecate TopicPartitionInitialOffset"

Previous commit caused a breaking API change.

Polish streams; move utility conversion methods to deprecated class

Switch to varargs in container factory

Code style polishing

Author: garyrussell

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

@@ -217,10 +217,10 @@ public void setApplicationEventPublisher(ApplicationEventPublisher applicationEv
 	public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 		if (replyTemplate instanceof ReplyingKafkaOperations) {
 			this.logger.warn(
-				"The 'replyTemplate' should not be an implementation of 'ReplyingKafkaOperations'; "
-				+ "such implementations are for client-side request/reply operations; here we "
-				+ "are simply sending a reply to an incoming request so the reply container will "
-				+ "never be used and will consume unnecessary resources.");
+					"The 'replyTemplate' should not be an implementation of 'ReplyingKafkaOperations'; "
+							+ "such implementations are for client-side request/reply operations; here we "
+							+ "are simply sending a reply to an incoming request so the reply container will "
+							+ "never be used and will consume unnecessary resources.");
 		}
 		this.replyTemplate = replyTemplate;
 	}
@@ -314,7 +314,7 @@ public void afterPropertiesSet() {
 	public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		C instance = createContainerInstance(endpoint);
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
+				.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
 		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
 			configureEndpoint((AbstractKafkaListenerEndpoint<K, V>) endpoint);
 		}
@@ -327,14 +327,14 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 
 	private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint) {
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(this.recordFilterStrategy, aklEndpoint::setRecordFilterStrategy)
-			.acceptIfNotNull(this.ackDiscarded, aklEndpoint::setAckDiscarded)
-			.acceptIfNotNull(this.retryTemplate, aklEndpoint::setRetryTemplate)
-			.acceptIfNotNull(this.recoveryCallback, aklEndpoint::setRecoveryCallback)
-			.acceptIfNotNull(this.statefulRetry, aklEndpoint::setStatefulRetry)
-			.acceptIfNotNull(this.batchListener, aklEndpoint::setBatchListener)
-			.acceptIfNotNull(this.replyTemplate, aklEndpoint::setReplyTemplate)
-			.acceptIfNotNull(this.replyHeadersConfigurer, aklEndpoint::setReplyHeadersConfigurer);
+				.acceptIfNotNull(this.recordFilterStrategy, aklEndpoint::setRecordFilterStrategy)
+				.acceptIfNotNull(this.ackDiscarded, aklEndpoint::setAckDiscarded)
+				.acceptIfNotNull(this.retryTemplate, aklEndpoint::setRetryTemplate)
+				.acceptIfNotNull(this.recoveryCallback, aklEndpoint::setRecoveryCallback)
+				.acceptIfNotNull(this.statefulRetry, aklEndpoint::setStatefulRetry)
+				.acceptIfNotNull(this.batchListener, aklEndpoint::setBatchListener)
+				.acceptIfNotNull(this.replyTemplate, aklEndpoint::setReplyTemplate)
+				.acceptIfNotNull(this.replyHeadersConfigurer, aklEndpoint::setReplyHeadersConfigurer);
 	}
 
 	/**
@@ -356,13 +356,13 @@ protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 		BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",
 				"messageListener", "ackCount", "ackTime");
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(this.afterRollbackProcessor, instance::setAfterRollbackProcessor)
-			.acceptIfCondition(this.containerProperties.getAckCount() > 0, this.containerProperties.getAckCount(),
-					properties::setAckCount)
-			.acceptIfCondition(this.containerProperties.getAckTime() > 0, this.containerProperties.getAckTime(),
-					properties::setAckTime)
-			.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
-			.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal);
+				.acceptIfNotNull(this.afterRollbackProcessor, instance::setAfterRollbackProcessor)
+				.acceptIfCondition(this.containerProperties.getAckCount() > 0, this.containerProperties.getAckCount(),
+						properties::setAckCount)
+				.acceptIfCondition(this.containerProperties.getAckTime() > 0, this.containerProperties.getAckTime(),
+						properties::setAckTime)
+				.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
+				.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal);
 		if (endpoint.getAutoStartup() != null) {
 			instance.setAutoStartup(endpoint.getAutoStartup());
 		}
@@ -371,54 +371,66 @@ else if (this.autoStartup != null) {
 		}
 		instance.setRecordInterceptor(this.recordInterceptor);
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(this.phase, instance::setPhase)
-			.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
-			.acceptIfNotNull(endpoint.getGroupId(), instance.getContainerProperties()::setGroupId)
-			.acceptIfNotNull(endpoint.getClientIdPrefix(), instance.getContainerProperties()::setClientId)
+				.acceptIfNotNull(this.phase, instance::setPhase)
+				.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
+				.acceptIfNotNull(endpoint.getGroupId(), instance.getContainerProperties()::setGroupId)
+				.acceptIfNotNull(endpoint.getClientIdPrefix(), instance.getContainerProperties()::setClientId)
 				.acceptIfNotNull(endpoint.getConsumerProperties(),
 						instance.getContainerProperties()::setConsumerProperties);
 	}
 
+	/**
+	 * {@inheritDoc}
+	 * @deprecated in favor of {@link #createContainer(TopicPartitionOffset[])}
+	 */
+	@Deprecated
+	@Override
+	public C createContainer(Collection<org.springframework.kafka.support.TopicPartitionInitialOffset> topicPartitions) {
+		return createContainer(topicPartitions.stream()
+				.map(org.springframework.kafka.support.TopicPartitionInitialOffset::toTPO)
+				.toArray(TopicPartitionOffset[]::new));
+	}
+
 	@Override
-	public C createContainer(final Collection<TopicPartitionOffset> topicPartitions) {
+	public C createContainer(TopicPartitionOffset... topicsAndPartitions) {
 		KafkaListenerEndpoint endpoint = new KafkaListenerEndpointAdapter() {
 
-					@Override
-					public Collection<TopicPartitionOffset> getTopicPartitions() {
-						return topicPartitions;
-					}
+			@Override
+			public TopicPartitionOffset[] getTopicPartitionsToAssign() {
+				return Arrays.copyOf(topicsAndPartitions, topicsAndPartitions.length);
+			}
 
-				};
+		};
 		C container = createContainerInstance(endpoint);
 		initializeContainer(container, endpoint);
 		return container;
 	}
 
 	@Override
-	public C createContainer(final String... topics) {
+	public C createContainer(String... topics) {
 		KafkaListenerEndpoint endpoint = new KafkaListenerEndpointAdapter() {
 
-					@Override
-					public Collection<String> getTopics() {
-						return Arrays.asList(topics);
-					}
+			@Override
+			public Collection<String> getTopics() {
+				return Arrays.asList(topics);
+			}
 
-				};
+		};
 		C container = createContainerInstance(endpoint);
 		initializeContainer(container, endpoint);
 		return container;
 	}
 
 	@Override
-	public C createContainer(final Pattern topicPattern) {
+	public C createContainer(Pattern topicPattern) {
 		KafkaListenerEndpoint endpoint = new KafkaListenerEndpointAdapter() {
 
-					@Override
-					public Pattern getTopicPattern() {
-						return topicPattern;
-					}
+			@Override
+			public Pattern getTopicPattern() {
+				return topicPattern;
+			}
 
-				};
+		};
 		C container = createContainerInstance(endpoint);
 		initializeContainer(container, endpoint);
 		return container;

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

@@ -51,6 +51,7 @@
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Base model for a Kafka listener endpoint.
@@ -190,6 +191,25 @@ public Collection<String> getTopics() {
 	 * Either this or 'topic' or 'topicPattern'
 	 * should be provided, but not a mixture.
 	 * @param topicPartitions to set.
+	 * @deprecated in favor of {@link #setTopicPartitions(TopicPartitionOffset...)}.
+	 * @see #setTopics(String...)
+	 * @see #setTopicPattern(Pattern)
+	 */
+	@Deprecated
+	public void setTopicPartitions(org.springframework.kafka.support.TopicPartitionInitialOffset... topicPartitions) {
+		Assert.notNull(topicPartitions, "'topics' must not be null");
+		this.topicPartitions.clear();
+		Arrays.stream(topicPartitions)
+				.map(org.springframework.kafka.support.TopicPartitionInitialOffset::toTPO)
+				.forEach(this.topicPartitions::add);
+	}
+
+	/**
+	 * Set the topicPartitions to use.
+	 * Either this or 'topic' or 'topicPattern'
+	 * should be provided, but not a mixture.
+	 * @param topicPartitions to set.
+	 * @since 2.3
 	 * @see #setTopics(String...)
 	 * @see #setTopicPattern(Pattern)
 	 */
@@ -202,10 +222,11 @@ public void setTopicPartitions(TopicPartitionOffset... topicPartitions) {
 	/**
 	 * Return the topicPartitions for this endpoint.
 	 * @return the topicPartitions for this endpoint.
+	 * @since 2.3
 	 */
 	@Override
-	public Collection<TopicPartitionOffset> getTopicPartitions() {
-		return Collections.unmodifiableCollection(this.topicPartitions);
+	public TopicPartitionOffset[] getTopicPartitionsToAssign() {
+		return this.topicPartitions.toArray(new TopicPartitionOffset[0]);
 	}
 
 	/**
@@ -414,7 +435,7 @@ public void setConsumerProperties(Properties consumerProperties) {
 	@Override
 	public void afterPropertiesSet() {
 		boolean topicsEmpty = getTopics().isEmpty();
-		boolean topicPartitionsEmpty = getTopicPartitions().isEmpty();
+		boolean topicPartitionsEmpty = ObjectUtils.isEmpty(getTopicPartitionsToAssign());
 		if (!topicsEmpty && !topicPartitionsEmpty) {
 			throw new IllegalStateException("Topics or topicPartitions must be provided but not both for " + this);
 		}
@@ -459,11 +480,11 @@ private void setupMessageListener(MessageListenerContainer container, MessageCon
 			if (this.batchListener) {
 				if (((MessagingMessageListenerAdapter<K, V>) messageListener).isConsumerRecords()) {
 					this.logger.warn(() -> "Filter strategy ignored when consuming 'ConsumerRecords'"
-								+ (this.id != null ? " id: " + this.id : ""));
+							+ (this.id != null ? " id: " + this.id : ""));
 				}
 				else {
 					messageListener = new FilteringBatchMessageListenerAdapter<>(
-						(BatchMessageListener<K, V>) messageListener, this.recordFilterStrategy, this.ackDiscarded);
+							(BatchMessageListener<K, V>) messageListener, this.recordFilterStrategy, this.ackDiscarded);
 				}
 			}
 			else {

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

@@ -59,27 +59,28 @@ public void setConcurrency(Integer concurrency) {
 
 	@Override
 	protected ConcurrentMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
-		Collection<TopicPartitionOffset> topicPartitions = endpoint.getTopicPartitions();
-		if (!topicPartitions.isEmpty()) {
-			ContainerProperties properties = new ContainerProperties(
-					topicPartitions.toArray(new TopicPartitionOffset[0]));
-			return new ConcurrentMessageListenerContainer<K, V>(getConsumerFactory(), properties);
+		TopicPartitionOffset[] topicPartitions = endpoint.getTopicPartitionsToAssign();
+		if (topicPartitions != null && topicPartitions.length > 0) {
+			ContainerProperties properties = new ContainerProperties(topicPartitions);
+			return new ConcurrentMessageListenerContainer<>(getConsumerFactory(), properties);
 		}
 		else {
 			Collection<String> topics = endpoint.getTopics();
 			if (!topics.isEmpty()) {
 				ContainerProperties properties = new ContainerProperties(topics.toArray(new String[0]));
-				return new ConcurrentMessageListenerContainer<K, V>(getConsumerFactory(), properties);
+				return new ConcurrentMessageListenerContainer<>(getConsumerFactory(), properties);
 			}
 			else {
 				ContainerProperties properties = new ContainerProperties(endpoint.getTopicPattern());
-				return new ConcurrentMessageListenerContainer<K, V>(getConsumerFactory(), properties);
+				return new ConcurrentMessageListenerContainer<>(getConsumerFactory(), properties);
 			}
 		}
 	}
 
 	@Override
-	protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> instance, KafkaListenerEndpoint endpoint) {
+	protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> instance,
+			KafkaListenerEndpoint endpoint) {
+
 		super.initializeContainer(instance, endpoint);
 		if (endpoint.getConcurrency() != null) {
 			instance.setConcurrency(endpoint.getConcurrency());

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

@@ -46,11 +46,23 @@ public interface KafkaListenerContainerFactory<C extends MessageListenerContaine
 	 * Create and configure a container without a listener; used to create containers that
 	 * are not used for KafkaListener annotations. Containers created using this method
 	 * are not added to the listener endpoint registry.
-	 * @param topicPartitions the topicPartitions.
+	 * @param topicPartitions the topicPartitions to assign.
+	 * @deprecated in favor of {@link #createContainer(TopicPartitionOffset[])}.
 	 * @return the container.
 	 * @since 2.2
 	 */
-	C createContainer(Collection<TopicPartitionOffset> topicPartitions);
+	@Deprecated
+	C createContainer(Collection<org.springframework.kafka.support.TopicPartitionInitialOffset> topicPartitions);
+
+	/**
+	 * Create and configure a container without a listener; used to create containers that
+	 * are not used for KafkaListener annotations. Containers created using this method
+	 * are not added to the listener endpoint registry.
+	 * @param topicPartitions the topicPartitions to assign.
+	 * @return the container.
+	 * @since 2.3
+	 */
+	C createContainer(TopicPartitionOffset... topicPartitions);
 
 	/**
 	 * Create and configure a container without a listener; used to create containers that

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

@@ -16,9 +16,12 @@
 
 package org.springframework.kafka.config;
 
+import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Properties;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -67,8 +70,21 @@ public interface KafkaListenerEndpoint {
 	/**
 	 * Return the topicPartitions for this endpoint.
 	 * @return the topicPartitions for this endpoint.
+	 * @deprecated in favor of {@link #getTopicPartitionsToAssign()}.
 	 */
-	Collection<TopicPartitionOffset> getTopicPartitions();
+	@Deprecated
+	default Collection<org.springframework.kafka.support.TopicPartitionInitialOffset> getTopicPartitions() {
+		return Arrays.stream(getTopicPartitionsToAssign())
+				.map(org.springframework.kafka.support.TopicPartitionInitialOffset::fromTPO)
+				.collect(Collectors.collectingAndThen(Collectors.toList(), Collections::unmodifiableList));
+	}
+
+	/**
+	 * Return the topicPartitions for this endpoint.
+	 * @return the topicPartitions for this endpoint.
+	 * @since 2.3
+	 */
+	TopicPartitionOffset[] getTopicPartitionsToAssign();
 
 	/**
 	 * Return the topicPattern for this endpoint.

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

@@ -58,8 +58,8 @@ public Collection<String> getTopics() {
 	}
 
 	@Override
-	public Collection<TopicPartitionOffset> getTopicPartitions() {
-		return Collections.emptyList();
+	public TopicPartitionOffset[] getTopicPartitionsToAssign() {
+		return new TopicPartitionOffset[0];
 	}
 
 	@Override