From 2c3f26e905aa7675df7e011ea5803b5810595030 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 20 Oct 2020 13:12:22 -0400 Subject: [PATCH] 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** --- .../AbstractMessageListenerContainer.java | 33 +++++++++++++++++++ .../DirectMessageListenerContainer.java | 4 +++ .../SimpleMessageListenerContainer.java | 3 ++ src/reference/asciidoc/amqp.adoc | 9 +++++ src/reference/asciidoc/whats-new.adoc | 3 ++ 5 files changed, 52 insertions(+) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index d05516c8..a587257b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/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. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index 4061af0f..d270132d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/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) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index 5ecc81ea..6f7f78d1 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/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; } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 3252cf6a..15d8fadc 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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) diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 7313332c..df45b7da 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -37,6 +37,9 @@ See <> 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 <> for more information. ==== MessagePostProcessor Changes