GH-1970: Container Stopping EH Improvement

Add an option to stop the container normally.

There might be a case where the container is stopped due to an error, but
the user wants the container to remain 'healthy', e.g. to avoid Kubernetes
restarting the instance.

Author: garyrussell

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

@@ -40,6 +40,8 @@ public class CommonContainerStoppingErrorHandler extends KafkaExceptionLogLevelA
 
 	private final Executor executor;
 
+	private boolean stopContainerAbnormally = true;
+
 	/**
 	 * Construct an instance with a default {@link SimpleAsyncTaskExecutor}.
 	 */
@@ -56,6 +58,18 @@ public CommonContainerStoppingErrorHandler(Executor executor) {
 		this.executor = executor;
 	}
 
+	/**
+	 * Set to false to stop the container normally. By default, the container is stopped
+	 * abnormally, so that {@code container.isInExpectedState()} returns false. If you
+	 * want to container to remain "healthy" when using this error handler, set the
+	 * property to false.
+	 * @param stopContainerAbnormally false for normal stop.
+	 * @since 2.8
+	 */
+	public void setStopContainerAbnormally(boolean stopContainerAbnormally) {
+		this.stopContainerAbnormally = stopContainerAbnormally;
+	}
+
 	@Override
 	public boolean remainingRecords() {
 		return true;
@@ -84,8 +98,16 @@ public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, C
 	}
 
 	private void stopContainer(MessageListenerContainer container, Exception thrownException) {
-		this.executor.execute(() -> container.stopAbnormally(() -> {
-		}));
+		this.executor.execute(() -> {
+			if (this.stopContainerAbnormally) {
+				container.stopAbnormally(() -> {
+				});
+			}
+			else {
+				container.stop(() -> {
+				});
+			}
+		});
 		// isRunning is false before the container.stop() waits for listener thread
 		try {
 			ListenerUtils.stoppableSleep(container, 10_000); // NOSONAR

spring-kafka/src/test/java/org/springframework/kafka/listener/CommonContainerStoppingErrorHandler2Tests.java

@@ -104,6 +104,7 @@ public void stopContainerAfterException() throws Exception {
 		assertThat(this.config.count).isEqualTo(4);
 		assertThat(this.config.contents.toArray()).isEqualTo(new String[]
 				{ "foo", "bar", "baz", "qux" });
+		assertThat(container.isInExpectedState()).isTrue();
 	}
 
 	@Configuration
@@ -202,6 +203,8 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 				@Override
 				public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
 						Consumer<?, ?> consumer, MessageListenerContainer container) {
+
+					setStopContainerAbnormally(false);
 					RuntimeException exception = null;
 					try {
 						super.handleRemaining(thrownException, records, consumer, container);