Revert "Temporarily Restore Deprecations Affecting SIK"

This reverts commit bbf26bc.

Author: garyrussell

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

@@ -46,8 +46,6 @@
 import org.springframework.kafka.support.JavaUtils;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.converter.MessageConverter;
-import org.springframework.retry.RecoveryCallback;
-import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 
 /**
@@ -71,6 +69,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null); // NOSONAR
 
+	@SuppressWarnings("deprecation")
 	private org.springframework.kafka.listener.GenericErrorHandler<?> errorHandler;
 
 	private CommonErrorHandler commonErrorHandler;
@@ -87,12 +86,6 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private Boolean ackDiscarded;
 
-	private RetryTemplate retryTemplate;
-
-	private RecoveryCallback<? extends Object> recoveryCallback;
-
-	private Boolean statefulRetry;
-
 	private Boolean batchListener;
 
 	private ApplicationEventPublisher applicationEventPublisher;
@@ -174,43 +167,6 @@ public void setAckDiscarded(Boolean ackDiscarded) {
 		this.ackDiscarded = ackDiscarded;
 	}
 
-	/**
-	 * Set a retryTemplate.
-	 * @param retryTemplate the template.
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	public void setRetryTemplate(RetryTemplate retryTemplate) {
-		this.retryTemplate = retryTemplate;
-	}
-
-	/**
-	 * Set a callback to be used with the {@link #setRetryTemplate(RetryTemplate)
-	 * retryTemplate}.
-	 * @param recoveryCallback the callback.
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
-		this.recoveryCallback = recoveryCallback;
-	}
-
-	/**
-	 * When using a {@link RetryTemplate} Set to true to enable stateful retry. Use in
-	 * conjunction with a legacy
-	 * {@code org.springframework.kafka.listener.SeekToCurrentErrorHandler} when retry can
-	 * take excessive time; each failure goes back to the broker, to keep the Consumer
-	 * alive.
-	 * @param statefulRetry true to enable stateful retry.
-	 * @since 2.1.3
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	public void setStatefulRetry(boolean statefulRetry) {
-		this.statefulRetry = statefulRetry;
-	}
-
-
 	/**
 	 * Return true if this endpoint creates a batch listener.
 	 * @return true for a batch listener.
@@ -398,17 +354,13 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		return instance;
 	}
 
-	@SuppressWarnings("deprecation")
 	private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint) {
 		if (aklEndpoint.getRecordFilterStrategy() == null) {
 			JavaUtils.INSTANCE
 					.acceptIfNotNull(this.recordFilterStrategy, aklEndpoint::setRecordFilterStrategy);
 		}
 		JavaUtils.INSTANCE
 				.acceptIfNotNull(this.ackDiscarded, aklEndpoint::setAckDiscarded)
-				.acceptIfNotNull(this.retryTemplate, aklEndpoint::setRetryTemplate)
-				.acceptIfNotNull(this.recoveryCallback, aklEndpoint::setRecoveryCallback)
-				.acceptIfNotNull(this.statefulRetry, aklEndpoint::setStatefulRetry)
 				.acceptIfNotNull(this.replyTemplate, aklEndpoint::setReplyTemplate)
 				.acceptIfNotNull(this.replyHeadersConfigurer, aklEndpoint::setReplyHeadersConfigurer)
 				.acceptIfNotNull(this.batchToRecordAdapter, aklEndpoint::setBatchToRecordAdapter);

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

@@ -49,7 +49,6 @@
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.lang.Nullable;
 import org.springframework.retry.RecoveryCallback;
-import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
@@ -94,8 +93,6 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private boolean ackDiscarded;
 
-	private RetryTemplate retryTemplate;
-
 	private RecoveryCallback<? extends Object> recoveryCallback;
 
 	private boolean statefulRetry;
@@ -328,67 +325,6 @@ public void setAckDiscarded(boolean ackDiscarded) {
 		this.ackDiscarded = ackDiscarded;
 	}
 
-	@Nullable
-	protected RetryTemplate getRetryTemplate() {
-		return this.retryTemplate;
-	}
-
-	/**
-	 * Set a retryTemplate.
-	 * @param retryTemplate the template.
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	public void setRetryTemplate(RetryTemplate retryTemplate) {
-		this.retryTemplate = retryTemplate;
-	}
-
-	/**
-	 * Get the recovery callback.
-	 * @return the recovery callback.
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	@Nullable
-	protected RecoveryCallback<?> getRecoveryCallback() {
-		return this.recoveryCallback;
-	}
-
-	/**
-	 * Set a callback to be used with the {@link #setRetryTemplate(RetryTemplate)}.
-	 * @param recoveryCallback the callback.
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
-		this.recoveryCallback = recoveryCallback;
-	}
-
-	/**
-	 * Return the stateful retry.
-	 * @return the stateful retry.
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	protected boolean isStatefulRetry() {
-		return this.statefulRetry;
-	}
-
-	/**
-	 * When using a {@link RetryTemplate}, set to true to enable stateful retry. Use in
-	 * conjunction with a legacy
-	 * {@code org.springframework.kafka.listener.SeekToCurrentErrorHandler} when retry can
-	 * take excessive time; each failure goes back to the broker, to keep the Consumer
-	 * alive.
-	 * @param statefulRetry true to enable stateful retry.
-	 * @since 2.1.3
-	 * @deprecated since 2.8 - use a suitably configured error handler instead.
-	 */
-	@Deprecated
-	public void setStatefulRetry(boolean statefulRetry) {
-		this.statefulRetry = statefulRetry;
-	}
-
 	@Nullable
 	@Override
 	public String getClientIdPrefix() {
@@ -544,7 +480,7 @@ public void setupListenerContainer(MessageListenerContainer listenerContainer,
 	protected abstract MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
 			@Nullable MessageConverter messageConverter);
 
-	@SuppressWarnings({ "unchecked", "deprecation" })
+	@SuppressWarnings("unchecked")
 	private void setupMessageListener(MessageListenerContainer container,
 			@Nullable MessageConverter messageConverter) {
 
@@ -557,14 +493,6 @@ private void setupMessageListener(MessageListenerContainer container,
 		boolean isBatchListener = isBatchListener();
 		Assert.state(messageListener != null,
 				() -> "Endpoint [" + this + "] must provide a non null message listener");
-		Assert.state(this.retryTemplate == null || !isBatchListener,
-				"A 'RetryTemplate' is not supported with a batch listener; consider configuring the container "
-				+ "with a suitably configured 'SeekToCurrentBatchErrorHandler' instead");
-		if (this.retryTemplate != null) {
-			messageListener = new org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter<>(
-							(MessageListener<K, V>) messageListener,
-					this.retryTemplate, this.recoveryCallback, this.statefulRetry);
-		}
 		if (this.recordFilterStrategy != null) {
 			if (isBatchListener) {
 				if (((MessagingMessageListenerAdapter<K, V>) messageListener).isConsumerRecords()) {