spring-projectsGH-3661: Resolve Spring Apache Kafka Deprecations

Resolves spring-projects#3661

Author: garyrussell

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-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.
@@ -42,8 +42,8 @@
 import reactor.core.publisher.Flux;
 
 /**
- * A support class for producer endpoints that provides a setter for the
- * output channel and a convenience method for sending Messages.
+ * A support class for producer endpoints that provides a setter for the output channel
+ * and a convenience method for sending Messages.
  *
  * @author Mark Fisher
  * @author Artem Bilan
@@ -76,8 +76,8 @@ public void setOutputChannel(MessageChannel outputChannel) {
 	}
 
 	/**
-	 * Set the output channel name; overrides
-	 * {@link #setOutputChannel(MessageChannel) outputChannel} if provided.
+	 * Set the output channel name; overrides {@link #setOutputChannel(MessageChannel)
+	 * outputChannel} if provided.
 	 * @param outputChannelName the channel name.
 	 * @since 4.3
 	 */
@@ -114,8 +114,7 @@ public void setErrorChannelName(String errorChannelName) {
 	}
 
 	/**
-	 * Return the error channel (if provided) to which error messages will
-	 * be routed.
+	 * Return the error channel (if provided) to which error messages will be routed.
 	 * @return the channel or null.
 	 * @since 4.3
 	 */
@@ -130,8 +129,8 @@ public MessageChannel getErrorChannel() {
 	}
 
 	/**
-	 * Configure the default timeout value to use for send operations.
-	 * May be overridden for individual messages.
+	 * Configure the default timeout value to use for send operations. May be overridden
+	 * for individual messages.
 	 * @param sendTimeout the send timeout in milliseconds
 	 * @see MessagingTemplate#setSendTimeout
 	 */
@@ -145,8 +144,8 @@ public void setShouldTrack(boolean shouldTrack) {
 	}
 
 	/**
-	 * Set an {@link ErrorMessageStrategy} to use to build an error message when a exception occurs.
-	 * Default is the {@link DefaultErrorMessageStrategy}.
+	 * Set an {@link ErrorMessageStrategy} to use to build an error message when a
+	 * exception occurs. Default is the {@link DefaultErrorMessageStrategy}.
 	 * @param errorMessageStrategy the {@link ErrorMessageStrategy}.
 	 * @since 4.3.10
 	 */
@@ -155,6 +154,16 @@ public final void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrat
 		this.errorMessageStrategy = errorMessageStrategy;
 	}
 
+	/**
+	 * Get an {@link ErrorMessageStrategy} to use to build an error message when a
+	 * exception occurs. Default is the {@link DefaultErrorMessageStrategy}.
+	 * @return the errorMessageStrategy
+	 * @since 6.0
+	 */
+	protected ErrorMessageStrategy getErrorMessageStrategy() {
+		return this.errorMessageStrategy;
+	}
+
 	protected MessagingTemplate getMessagingTemplate() {
 		return this.messagingTemplate;
 	}
@@ -181,18 +190,16 @@ protected void onInit() {
 	}
 
 	/**
-	 * Take no action by default.
-	 * Subclasses may override this if they
-	 * need lifecycle-managed behavior. Protected by 'lifecycleLock'.
+	 * Take no action by default. Subclasses may override this if they need
+	 * lifecycle-managed behavior. Protected by 'lifecycleLock'.
 	 */
 	@Override
 	protected void doStart() {
 	}
 
 	/**
-	 * Take no action by default.
-	 * Subclasses may override this if they
-	 * need lifecycle-managed behavior.
+	 * Take no action by default. Subclasses may override this if they need
+	 * lifecycle-managed behavior.
 	 */
 	@Override
 	protected void doStop() {
@@ -217,12 +224,11 @@ protected void sendMessage(Message<?> messageArg) {
 	protected void subscribeToPublisher(Publisher<? extends Message<?>> publisher) {
 		MessageChannel channelForSubscription = getRequiredOutputChannel();
 
-		Flux<? extends Message<?>> messageFlux =
-				Flux.from(publisher)
-						.map(this::trackMessageIfAny)
-						.doOnComplete(this::stop)
-						.doOnCancel(this::stop)
-						.takeWhile((message) -> isActive());
+		Flux<? extends Message<?>> messageFlux = Flux.from(publisher)
+				.map(this::trackMessageIfAny)
+				.doOnComplete(this::stop)
+				.doOnCancel(this::stop)
+				.takeWhile((message) -> isActive());
 
 		if (channelForSubscription instanceof ReactiveStreamsSubscribableChannel) {
 			((ReactiveStreamsSubscribableChannel) channelForSubscription).subscribeTo(messageFlux);
@@ -248,7 +254,8 @@ protected void subscribeToPublisher(Publisher<? extends Message<?>> publisher) {
 	 * @return true if the error channel is available and message sent.
 	 * @since 4.3.10
 	 */
-	protected final boolean sendErrorMessageIfNecessary(@Nullable Message<?> message, Exception exception) {
+	protected final boolean sendErrorMessageIfNecessary(@Nullable
+	Message<?> message, Exception exception) {
 		MessageChannel channel = getErrorChannel();
 		if (channel != null) {
 			this.messagingTemplate.send(channel, buildErrorMessage(message, exception));
@@ -265,7 +272,8 @@ protected final boolean sendErrorMessageIfNecessary(@Nullable Message<?> message
 	 * @return the error message.
 	 * @since 4.3.10
 	 */
-	protected final ErrorMessage buildErrorMessage(@Nullable Message<?> message, Exception exception) {
+	protected final ErrorMessage buildErrorMessage(@Nullable
+	Message<?> message, Exception exception) {
 		return this.errorMessageStrategy.buildErrorMessage(exception, getErrorMessageAttributes(message));
 	}
 
@@ -277,7 +285,8 @@ protected final ErrorMessage buildErrorMessage(@Nullable Message<?> message, Exc
 	 * @return the attributes.
 	 * @since 4.3.10
 	 */
-	protected AttributeAccessor getErrorMessageAttributes(@Nullable Message<?> message) {
+	protected AttributeAccessor getErrorMessageAttributes(@Nullable
+	Message<?> message) {
 		return ErrorMessageUtils.getAttributeAccessor(message, null);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-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.
@@ -299,8 +299,8 @@ public boolean isLoggingEnabled() {
 	}
 
 	/**
-	 * Set an {@link ErrorMessageStrategy} to use to build an error message when a exception occurs.
-	 * Default is the {@link DefaultErrorMessageStrategy}.
+	 * Set an {@link ErrorMessageStrategy} to use to build an error message when a
+	 * exception occurs. Default is the {@link DefaultErrorMessageStrategy}.
 	 * @param errorMessageStrategy the {@link ErrorMessageStrategy}.
 	 * @since 4.3.10
 	 */
@@ -309,6 +309,16 @@ public final void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrat
 		this.errorMessageStrategy = errorMessageStrategy;
 	}
 
+	/**
+	 * Get an {@link ErrorMessageStrategy} to use to build an error message when a
+	 * exception occurs. Default is the {@link DefaultErrorMessageStrategy}.
+	 * @return the errorMessageStrategy.
+	 * @since 6.0
+	 */
+	protected ErrorMessageStrategy getErrorMessageStrategy() {
+		return this.errorMessageStrategy;
+	}
+
 	@Override
 	public ManagementOverrides getOverrides() {
 		return this.managementOverrides;

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java

@@ -24,6 +24,7 @@
 import org.springframework.core.task.AsyncListenableTaskExecutor;
 import org.springframework.integration.dsl.IntegrationComponentSpec;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.listener.CommonErrorHandler;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -81,17 +82,15 @@ public KafkaMessageListenerContainerSpec<K, V> concurrency(int concurrency) {
 	}
 
 	/**
-	 * Specify an {@link org.springframework.kafka.listener.ErrorHandler} for the
+	 * Specify an {@link org.springframework.kafka.listener.CommonErrorHandler} for the
 	 * {@link org.springframework.kafka.listener.AbstractMessageListenerContainer}.
-	 * @param errorHandler the {@link org.springframework.kafka.listener.ErrorHandler}.
+	 * @param errorHandler the {@link org.springframework.kafka.listener.CommonErrorHandler}.
 	 * @return the spec.
-	 * @see org.springframework.kafka.listener.ErrorHandler
+	 * @since 6.0
+	 * @see org.springframework.kafka.listener.CommonErrorHandler
 	 */
-	@SuppressWarnings("deprecation")
-	public KafkaMessageListenerContainerSpec<K, V> errorHandler(
-			org.springframework.kafka.listener.GenericErrorHandler<?> errorHandler) {
-
-		this.target.setGenericErrorHandler(errorHandler);
+	public KafkaMessageListenerContainerSpec<K, V> errorHandler(CommonErrorHandler errorHandler) {
+		this.target.setCommonErrorHandler(errorHandler);
 		return this;
 	}
 

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaErrorSendingMessageRecoverer.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 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.
+ * 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.integration.kafka.inbound;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.core.AttributeAccessor;
+import org.springframework.integration.core.ErrorMessagePublisher;
+import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageUtils;
+import org.springframework.kafka.listener.ConsumerRecordRecoverer;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * An extension of {@link ErrorMessagePublisher} that can be used in a
+ * {@link org.springframework.kafka.listener.CommonErrorHandler} for recovering Kafka
+ * delivery failures.
+ *
+ * @author Gary Russell
+ * @since 6.0
+ *
+ */
+public class KafkaErrorSendingMessageRecoverer extends ErrorMessagePublisher implements ConsumerRecordRecoverer {
+
+	/**
+	 * Construct an instance to send to the channel with the
+	 * {@link RawRecordHeaderErrorMessageStrategy}.
+	 * @param channel the channel.
+	 */
+	public KafkaErrorSendingMessageRecoverer(MessageChannel channel) {
+		this(channel, new RawRecordHeaderErrorMessageStrategy());
+	}
+
+	/**
+	 * Construct an instance to send the channel, using the error message strategy.
+	 * @param channel the channel.
+	 * @param errorMessageStrategy the strategy.
+	 */
+	public KafkaErrorSendingMessageRecoverer(MessageChannel channel, ErrorMessageStrategy errorMessageStrategy) {
+		setChannel(channel);
+		setErrorMessageStrategy(errorMessageStrategy);
+	}
+
+	@Override
+	public void accept(ConsumerRecord<?, ?> record, Exception ex) {
+		Throwable thrown = ex.getCause();
+		if (thrown == null) {
+			thrown = ex;
+		}
+		AttributeAccessor attrs = ErrorMessageUtils.getAttributeAccessor(null, null);
+		attrs.setAttribute(KafkaHeaders.RAW_DATA, record);
+		publish(thrown, attrs);
+	}
+
+}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundEndpoint.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 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.
+ * 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.integration.kafka.inbound;
+
+import org.apache.kafka.clients.consumer.Consumer;
+
+import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.support.RetryTemplate;
+
+/**
+ * @author Gary Russell
+ * @since 6.0
+ *
+ */
+public interface KafkaInboundEndpoint {
+
+	/**
+	 * {@link org.springframework.retry.RetryContext} attribute key for an acknowledgment
+	 * if the listener is capable of acknowledging.
+	 */
+	String CONTEXT_ACKNOWLEDGMENT = "acknowledgment";
+
+	/**
+	 * {@link org.springframework.retry.RetryContext} attribute key for the consumer if
+	 * the listener is consumer-aware.
+	 */
+	String CONTEXT_CONSUMER = "consumer";
+
+	/**
+	 * {@link org.springframework.retry.RetryContext} attribute key for the record.
+	 */
+	String CONTEXT_RECORD = "record";
+
+	/**
+	 * Execute the runnable with the retry template and recovery callback.
+	 * @param template the template.
+	 * @param callback the callback.
+	 * @param record the record (or records).
+	 * @param acknowledgment the acknowledgment.
+	 * @param consumer the consumer.
+	 * @param runnable the runnable.
+	 */
+	default void doWithRetry(RetryTemplate template, RecoveryCallback<?> callback, Object data,
+			Acknowledgment acknowledgment, Consumer<?, ?> consumer, Runnable runnable) {
+
+		try {
+			template.execute(context -> {
+				context.setAttribute(CONTEXT_RECORD, data);
+				context.setAttribute(CONTEXT_ACKNOWLEDGMENT, acknowledgment);
+				context.setAttribute(CONTEXT_CONSUMER, consumer);
+				runnable.run();
+				return null;
+			}, callback);
+		}
+		catch (Exception ex) {
+			throw new KafkaException("Failed to execute runnable", ex);
+		}
+	}
+
+}