Fix MessageRecoverer.recover() for nullable cause

Author: artembilan

spring-amqp/src/main/java/org/springframework/amqp/AmqpRejectAndDontRequeueException.java

@@ -46,7 +46,7 @@ public AmqpRejectAndDontRequeueException(String message) {
 	 * Construct an instance with the supplied argument.
 	 * @param cause the cause.
 	 */
-	public AmqpRejectAndDontRequeueException(Throwable cause) {
+	public AmqpRejectAndDontRequeueException(@Nullable Throwable cause) {
 		this(null, false, cause);
 	}
 
@@ -55,7 +55,7 @@ public AmqpRejectAndDontRequeueException(Throwable cause) {
 	 * @param message A message describing the problem.
 	 * @param cause the cause.
 	 */
-	public AmqpRejectAndDontRequeueException(String message, Throwable cause) {
+	public AmqpRejectAndDontRequeueException(String message, @Nullable Throwable cause) {
 		this(message, false, cause);
 	}
 

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/StreamMessageRecoverer.java

@@ -17,6 +17,7 @@
 package org.springframework.rabbit.stream.retry;
 
 import com.rabbitmq.stream.MessageHandler.Context;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.retry.MessageRecoverer;
@@ -33,7 +34,7 @@
 public interface StreamMessageRecoverer extends MessageRecoverer {
 
 	@Override
-	default void recover(Message message, Throwable cause) {
+	default void recover(Message message, @Nullable Throwable cause) {
 	}
 
 	/**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/ImmediateRequeueMessageRecoverer.java

@@ -18,6 +18,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.ImmediateRequeueAmqpException;
 import org.springframework.amqp.core.Message;
@@ -36,11 +37,16 @@ public class ImmediateRequeueMessageRecoverer implements MessageRecoverer {
 	protected Log logger = LogFactory.getLog(ImmediateRequeueMessageRecoverer.class); // NOSONAR protected
 
 	@Override
-	public void recover(Message message, Throwable cause) {
+	public void recover(Message message, @Nullable Throwable cause) {
 		if (this.logger.isWarnEnabled()) {
 			this.logger.warn("Retries exhausted for message " + message + "; requeuing...", cause);
 		}
-		throw new ImmediateRequeueAmqpException(cause);
+		if (cause != null) {
+			throw new ImmediateRequeueAmqpException(cause);
+		}
+		else {
+			throw new ImmediateRequeueAmqpException("Re-queueing for message: " + message);
+		}
 	}
 
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/MessageBatchRecoverer.java

@@ -18,6 +18,8 @@
 
 import java.util.List;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.amqp.core.Message;
 
 /**
@@ -26,23 +28,24 @@
  * the recoverer to properly recover the remaining records.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.2
  *
  */
 @FunctionalInterface
 public interface MessageBatchRecoverer extends MessageRecoverer {
 
 	@Override
-	default void recover(Message message, Throwable cause) {
+	default void recover(Message message, @Nullable Throwable cause) {
 		throw new IllegalStateException("MessageBatchRecoverer configured with a non-batch listener");
 	}
 
 	/**
 	 * Callback for message batch that was consumed but failed all retry attempts.
-	 *
 	 * @param messages the messages to recover
 	 * @param cause the cause of the error
 	 */
-	void recover(List<Message> messages, Throwable cause);
+	void recover(List<Message> messages, @Nullable Throwable cause);
 
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/MessageRecoverer.java

@@ -16,6 +16,8 @@
 
 package org.springframework.amqp.rabbit.retry;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.amqp.core.Message;
 
 /**
@@ -24,6 +26,7 @@
  *
  * @author Dave Syer
  * @author Gary Russell
+ * @author Artem Bilan
  *
  */
 @FunctionalInterface
@@ -35,6 +38,6 @@ public interface MessageRecoverer {
 	 * @param message the message to recover
 	 * @param cause the cause of the error
 	 */
-	void recover(Message message, Throwable cause);
+	void recover(Message message, @Nullable Throwable cause);
 
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RejectAndDontRequeueRecoverer.java

@@ -20,6 +20,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.AmqpRejectAndDontRequeueException;
 import org.springframework.amqp.core.Message;
@@ -69,7 +70,7 @@ public RejectAndDontRequeueRecoverer(Supplier<String> messageSupplier) {
 	}
 
 	@Override
-	public void recover(Message message, Throwable cause) {
+	public void recover(Message message, @Nullable Throwable cause) {
 		if (this.logger.isWarnEnabled()) {
 			this.logger.warn("Retries exhausted for message " + message, cause);
 		}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java

@@ -189,18 +189,20 @@ protected MessageDeliveryMode getDeliveryMode() {
 	}
 
 	@Override
-	public void recover(Message message, Throwable cause) {
+	public void recover(Message message, @Nullable Throwable cause) {
 		MessageProperties messageProperties = message.getMessageProperties();
 		Map<String, @Nullable Object> headers = messageProperties.getHeaders();
-		String exceptionMessage = cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage();
-		@Nullable String[] processed = processStackTrace(cause, exceptionMessage);
-		String stackTraceAsString = processed[0];
-		String truncatedExceptionMessage = processed[1];
-		if (truncatedExceptionMessage != null) {
-			exceptionMessage = truncatedExceptionMessage;
+		if (cause != null) {
+			String exceptionMessage = cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage();
+			@Nullable String[] processed = processStackTrace(cause, exceptionMessage);
+			String stackTraceAsString = processed[0];
+			String truncatedExceptionMessage = processed[1];
+			if (truncatedExceptionMessage != null) {
+				exceptionMessage = truncatedExceptionMessage;
+			}
+			headers.put(X_EXCEPTION_STACKTRACE, stackTraceAsString);
+			headers.put(X_EXCEPTION_MESSAGE, exceptionMessage);
 		}
-		headers.put(X_EXCEPTION_STACKTRACE, stackTraceAsString);
-		headers.put(X_EXCEPTION_MESSAGE, exceptionMessage);
 		headers.put(X_ORIGINAL_EXCHANGE, messageProperties.getReceivedExchange());
 		headers.put(X_ORIGINAL_ROUTING_KEY, messageProperties.getReceivedRoutingKey());
 		Map<? extends String, ?> additionalHeaders = additionalHeaders(message, cause);
@@ -304,7 +306,7 @@ else if (stackTraceAsString.length() + exceptionMessage.length() > this.maxStack
 	 * @param cause The cause.
 	 * @return A {@link Map} of additional headers to add.
 	 */
-	protected @Nullable Map<? extends String, ?> additionalHeaders(Message message, Throwable cause) {
+	protected @Nullable Map<? extends String, ?> additionalHeaders(Message message, @Nullable Throwable cause) {
 		return null;
 	}