ListenerConsumer doesn't handle JVM Error exception.

[ingvard]

Hello,
Need your advice. In my application, I make a simple health check for a listener container.

    protected void doHealthCheck(Health.Builder builder) {
        for (MessageListenerContainer container : listenerRegistry.getListenerContainers()) {
            if (!container.isRunning()) {
                builder.down();
                return;
            }
        }

        builder.up();
    }

But it doesn't work as expected, because if a listener thread throws an error exception like OutOfMemoryException (Type: Error), the exception isn't caught and the thread dies. The running flag would be true for a dead container.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Line 714 in 7b214f8

while (isRunning()) {

I made a workaround, like -XX:+CrashOnOutOfMemoryError -XX:+ExitOnOutOfMemoryError.
I understand that catching a throwable is a bad practice, but could you catch it and change the flag?:)

[artembilan]

I think this is good idea to catch that respectively.

We have something like this in Spring AMQP listener container:

catch (Error e) { //NOSONAR
	// ok to catch Error - we're aborting so will stop
	logger.error("Consumer thread error, thread abort.", e);
	publishConsumerFailedEvent("Consumer threw an Error", true, e);
	aborted = true;
}

Feel free to proceed with the contribution on the matter!

Thank you @ingvard for the report!

[garyrussell]

Unfortunately, it's not that simple - the ConcurrentMLC instantiates multiple KMLCs. Just setting the flag on the child container won't be reflected on the CMLC's running flag.

And, what to do if some child containers are still running?

I suppose we should stop() the outer container when such an error occurs but that might fail in an OOM situation (although stop() doesn't create any new objects so it might be ok).

To avoid a tangle between the two classes we'd have to use some kind of callback to invoke the outer container's stop().

[artembilan]

Runnable stopCallbackFatal ? And call it when we catch that Error in any child listener container.
So, the parent one can iterate over all its children and stop them respectively.

[garyrussell]

Correct.