Shutdown MessageListenerContainer after max retries exceeded

[m-kay]

Can we somehow configure the DefaultErrorHandler to shutdown the MessageListenerContainer after the max number of retries exceeded instead of discarding the failed message? I already tried using the CommonContainerStoppingErrorHandler but this errorhandler stops immediately if an error occurs.

[garyrussell]

Simply add a custom recoverer to a subclass of the DefaultErrorHandler that calls the CommonContainerStoppingErrorHandler when it is called after the retries are exhausted. Something like this should work:

@Bean
CommonErrorHandler eh() {
	CommonContainerStoppingErrorHandler cseh = new CommonContainerStoppingErrorHandler();
	AtomicReference<Consumer<? ,?>> consumer2 = new AtomicReference<>();
	AtomicReference<MessageListenerContainer> container2 = new AtomicReference<>();

	return new DefaultErrorHandler((rec, ex) -> {
		cseh.handleRemaining(ex, Collections.singletonList(rec), consumer2.get(), container2.get());
	}, new FixedBackOff(1000L, 4)) {

		@Override
		public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
				Consumer<?, ?> consumer, MessageListenerContainer container) {

			consumer2.set(consumer);
			container2.set(container);
			super.handleRemaining(thrownException, records, consumer, container);
		}


	};
}

[m-kay]

Thank you, that works like a charm!

I could even simplify it because I'm configuring the errorhandler in a listener customizer and therefore already have a reference to the container.

@Bean
fun listenerContainerCustomizer(): ListenerContainerCustomizer<AbstractMessageListenerContainer<Any, Any>> {
    return ListenerContainerCustomizer { container, _, _ ->
        val containerStoppingErrorHandler = CommonContainerStoppingErrorHandler()
        container.commonErrorHandler = DefaultErrorHandler({ _: ConsumerRecord<*, *>, ex: Exception ->
            containerStoppingErrorHandler.handleRemaining(
                ex, listOf(), MockConsumer<String, String>(OffsetResetStrategy.NONE),
                container
            )
        }, SeekUtils.DEFAULT_BACK_OFF)
    }
}