GH-2089: Remove 2.x Deprecations

Resolves #2089

- remove deprecations
- remove implementations of legacy error handlers
- deprecate legacy error handler interfaces

* Fix deprecation warnings.

* Fix more deprecation warnings.

Author: garyrussell

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

@@ -6,9 +6,9 @@ For changes in earlier version, see <<history>>.
 [[x28-kafka-client]]
 ==== Kafka Client Version
 
-This version requires the 3.0.0 `kafka-clients`
+This version requires the 3.1.0 `kafka-clients`
 
-IMPORTANT: When using transactions, `kafka-clients` 3.0.0 and later no longer support `EOSMode.V2` (aka `BETA`) (and automatic fallback to `V1` - aka `ALPHA`) with brokers earlier than 2.5; you must therefore override the default `EOSMode` (`V2`) with `V1` if your brokers are older (or upgrade your brokers).
+IMPORTANT: When using transactions, the minimum broker version is 2.5.
 
 See <<exactly-once>> and https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
 

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

@@ -35,12 +35,9 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.kafka.listener.AfterRollbackProcessor;
-import org.springframework.kafka.listener.BatchErrorHandler;
 import org.springframework.kafka.listener.BatchInterceptor;
 import org.springframework.kafka.listener.CommonErrorHandler;
 import org.springframework.kafka.listener.ContainerProperties;
-import org.springframework.kafka.listener.ErrorHandler;
-import org.springframework.kafka.listener.GenericErrorHandler;
 import org.springframework.kafka.listener.RecordInterceptor;
 import org.springframework.kafka.listener.adapter.BatchToRecordAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
@@ -49,7 +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;
 
@@ -74,7 +70,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null); // NOSONAR
 
-	private GenericErrorHandler<?> errorHandler;
+	@SuppressWarnings("deprecation")
+	private org.springframework.kafka.listener.GenericErrorHandler<?> errorHandler;
 
 	private CommonErrorHandler commonErrorHandler;
 
@@ -90,10 +87,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;
@@ -177,25 +170,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.
-	 */
-	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
@@ -257,7 +231,7 @@ public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
 	@Deprecated
-	public void setErrorHandler(ErrorHandler errorHandler) {
+	public void setErrorHandler(org.springframework.kafka.listener.ErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
 
@@ -269,13 +243,14 @@ public void setErrorHandler(ErrorHandler errorHandler) {
 	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
 	@Deprecated
-	public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
+	public void setBatchErrorHandler(org.springframework.kafka.listener.BatchErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
 
 	/**
-	 * Set the {@link CommonErrorHandler} which can handle errors for both record
-	 * and batch listeners. Replaces the use of {@link GenericErrorHandler}s.
+	 * Set the {@link CommonErrorHandler} which can handle errors for both record and
+	 * batch listeners. Replaces the use of
+	 * {@link org.springframework.kafka.listener.GenericErrorHandler}s.
 	 * @param commonErrorHandler the handler.
 	 * @since 2.8
 	 */
@@ -361,16 +336,17 @@ public void setContainerCustomizer(ContainerCustomizer<K, V, C> containerCustomi
 		this.containerCustomizer = containerCustomizer;
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	public void afterPropertiesSet() {
 		if (this.commonErrorHandler == null && this.errorHandler != null) {
 			if (Boolean.TRUE.equals(this.batchListener)) {
-				Assert.state(this.errorHandler instanceof BatchErrorHandler,
+				Assert.state(this.errorHandler instanceof org.springframework.kafka.listener.BatchErrorHandler,
 						() -> "The error handler must be a BatchErrorHandler, not " +
 								this.errorHandler.getClass().getName());
 			}
 			else {
-				Assert.state(this.errorHandler instanceof ErrorHandler,
+				Assert.state(this.errorHandler instanceof org.springframework.kafka.listener.ErrorHandler,
 						() -> "The error handler must be an ErrorHandler, not " +
 								this.errorHandler.getClass().getName());
 			}
@@ -398,8 +374,6 @@ 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.replyTemplate, aklEndpoint::setReplyTemplate)
 				.acceptIfNotNull(this.replyHeadersConfigurer, aklEndpoint::setReplyHeadersConfigurer)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-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.
@@ -94,8 +94,6 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private boolean ackDiscarded;
 
-	private RetryTemplate retryTemplate;
-
 	private RecoveryCallback<? extends Object> recoveryCallback;
 
 	private boolean statefulRetry;
@@ -326,34 +324,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;
-	}
-
-	@Nullable
-	protected RecoveryCallback<?> getRecoveryCallback() {
-		return this.recoveryCallback;
-	}
-
-	/**
-	 * Set a callback to be used with the {@link #setRetryTemplate(RetryTemplate)}.
-	 * @param recoveryCallback the callback.
-	 */
-	public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
-		this.recoveryCallback = recoveryCallback;
-	}
-
 	protected boolean isStatefulRetry() {
 		return this.statefulRetry;
 	}
@@ -510,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) {
 
@@ -523,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()) {

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.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.
@@ -89,8 +89,6 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private StateRestoreListener stateRestoreListener;
 
-	private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
-
 	private StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler;
 
 	private boolean autoStartup = true;
@@ -191,17 +189,6 @@ public void setStateListener(KafkaStreams.StateListener stateListener) {
 		this.stateListener = stateListener; // NOSONAR (sync)
 	}
 
-	/**
-	 * Obsolete.
-	 * @param exceptionHandler the handler.
-	 * @deprecated in favor of
-	 * {@link #setStreamsUncaughtExceptionHandler(StreamsUncaughtExceptionHandler)}.
-	 */
-	@Deprecated
-	public void setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler exceptionHandler) {
-		this.uncaughtExceptionHandler = exceptionHandler; // NOSONAR (sync)
-	}
-
 	/**
 	 * Set a {@link StreamsUncaughtExceptionHandler}. Supercedes
 	 * {@link #setUncaughtExceptionHandler(java.lang.Thread.UncaughtExceptionHandler)}.
@@ -340,9 +327,6 @@ public synchronized void start() {
 				if (this.streamsUncaughtExceptionHandler != null) {
 					this.kafkaStreams.setUncaughtExceptionHandler(this.streamsUncaughtExceptionHandler);
 				}
-				else {
-					this.kafkaStreams.setUncaughtExceptionHandler(this.uncaughtExceptionHandler);
-				}
 				if (this.kafkaStreamsCustomizer != null) {
 					this.kafkaStreamsCustomizer.customize(this.kafkaStreams);
 				}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-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.
@@ -198,35 +198,6 @@ public interface KafkaOperations<K, V> {
 	 */
 	void flush();
 
-	/**
-	 * When running in a transaction, send the consumer offset(s) to the transaction. The
-	 * group id is obtained from
-	 * {@link org.springframework.kafka.support.KafkaUtils#getConsumerGroupId()}. It is
-	 * not necessary to call this method if the operations are invoked on a listener
-	 * container thread (and the listener container is configured with a
-	 * {@link org.springframework.kafka.transaction.KafkaAwareTransactionManager}) since
-	 * the container will take care of sending the offsets to the transaction.
-	 * @param offsets The offsets.
-	 * @since 1.3
-	 * @deprecated in the 3.0.0 KafkaProducer.
-	 */
-	@Deprecated
-	void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets);
-
-	/**
-	 * When running in a transaction, send the consumer offset(s) to the transaction. It
-	 * is not necessary to call this method if the operations are invoked on a listener
-	 * container thread (and the listener container is configured with a
-	 * {@link org.springframework.kafka.transaction.KafkaAwareTransactionManager}) since
-	 * the container will take care of sending the offsets to the transaction.
-	 * @param offsets The offsets.
-	 * @param consumerGroupId the consumer's group.id.
-	 * @since 1.3
-	 * @deprecated in the 3.0.0 KafkaProducer.
-	 */
-	@Deprecated
-	void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId);
-
 	/**
 	 * When running in a transaction, send the consumer offset(s) to the transaction. It
 	 * is not necessary to call this method if the operations are invoked on a listener

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-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.
@@ -546,20 +546,6 @@ public void flush() {
 		}
 	}
 
-
-	@Override
-	@Deprecated
-	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets) {
-		sendOffsetsToTransaction(offsets, KafkaUtils.getConsumerGroupId());
-	}
-
-	@SuppressWarnings("deprecation")
-	@Override
-	@Deprecated
-	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
-		producerForOffsets().sendOffsetsToTransaction(offsets, consumerGroupId);
-	}
-
 	@Override
 	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
 			ConsumerGroupMetadata groupMetadata) {

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-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.
@@ -102,18 +102,6 @@ public <T> T executeInTransaction(OperationsCallback<Object, Object, T> callback
 		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
 	}
 
-	@Override
-	@Deprecated
-	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
-		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
-	}
-
-	@Override
-	@Deprecated
-	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets) {
-		throw new UnsupportedOperationException(THIS_METHOD_IS_NOT_SUPPORTED);
-	}
-
 	@Override
 	public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
 			ConsumerGroupMetadata groupMetadata) {