GH-1258: Add OOMHandler to the listener containers

Resolves https://github.com/spring-projects/spring-amqp/issues/1258#issuecomment-711670729

**I will backport to 2.2.x with a no-op default**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java

@@ -246,6 +246,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	private long consumeDelay;
 
+	private OOMHandler oOMHandler = error -> System.exit(99);
+
 	private volatile boolean lazyLoad;
 
 	@Override
@@ -1145,6 +1147,21 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		this.consumeDelay = consumeDelay;
 	}
 
+	protected OOMHandler getOOMHandler() {
+		return this.oOMHandler;
+	}
+
+	/**
+	 * Provide an OOMHandler implementation; by default, {@code System.exit(99)} is
+	 * called.
+	 * @param oOMHandler the handler.
+	 * @since 2.2.12
+	 */
+	public void setOOMHandler(OOMHandler oOMHandler) {
+		Assert.notNull(oOMHandler, "'oOMHandler' cannot be null");
+		this.oOMHandler = oOMHandler;
+	}
+
 	/**
 	 * Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
 	 */
@@ -1972,6 +1989,22 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 
 	}
 
+	/**
+	 * A handler for {@link OutOfMemoryError} on the container thread(s).
+	 * @since 2.2.12
+	 *
+	 */
+	@FunctionalInterface
+	public interface OOMHandler {
+
+		/**
+		 * Handle the error; typically, the JVM will be terminated.
+		 * @param error the error.
+		 */
+		void handle(OutOfMemoryError error);
+
+	}
+
 	/**
 	 * Exception that indicates that the initial setup of this container's shared Rabbit Connection failed. This is
 	 * indicating to invokers that they need to establish the shared Connection themselves on first access.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java

@@ -1093,6 +1093,10 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 					}
 				}
 			}
+			catch (OutOfMemoryError e) { // NOSONAR
+				getOOMHandler().handle(e);
+				throw e;
+			}
 		}
 
 		private void handleAck(long deliveryTag, boolean channelLocallyTransacted) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java

@@ -1255,6 +1255,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 			catch (Error e) { //NOSONAR
 				// ok to catch Error - we're aborting so will stop
 				logger.error("Consumer thread error, thread abort.", e);
+				if (e instanceof OutOfMemoryError) { // NOSONAR
+					getOOMHandler().handle((OutOfMemoryError) e);
+				}
 				publishConsumerFailedEvent("Consumer threw an Error", true, e);
 				aborted = true;
 			}

src/reference/asciidoc/amqp.adoc

@@ -5789,6 +5789,15 @@ a|image::images/tickmark.png[]
 a|image::images/tickmark.png[]
 a|image::images/tickmark.png[]
 
+|oOMHandler
+(N/A)
+
+|An `AbstractMessageListenerContainer.OOMHandler` implementation that is called when a container thread catches an `OutOfMemoryException`.
+The default implementation calls `System.exit(99)`.
+
+a|image::images/tickmark.png[]
+a|image::images/tickmark.png[]
+
 |phase
 (phase)
 

src/reference/asciidoc/whats-new.adoc

@@ -37,6 +37,9 @@ See <<template-confirms>> for more information.
 ==== Listener Container Changes
 
 A new listener container property `consumeDelay` is now available; it is helpful when using the https://github.com/rabbitmq/rabbitmq-sharding[RabbitMQ Sharding Plugin].
+
+The default `OOMHandler` (out of memory handler) now calls `System.exit(99)`.
+
 See <<containerAttributes>> for more information.
 
 ==== MessagePostProcessor Changes