GH-1612: Option: Producer Fenced: Stop Container

Resolves #1612

**cherry-pick to 2.5.x**

* * Add @SInCE to javadocs; retain route cause of `StopAfterFenceException`.

* * Add reason to `ConsumerStoppedEvent`.

Resolves #1618

Also provide access to the actual container that stopped the consumer, for
example to allow restarting after stopping due to a producer fenced exception.

* * Add `@Nullable`s.

* * Test Polishing.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStoppedEvent.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.event;
 
+import org.springframework.lang.Nullable;
+
 /**
  * An event published when a consumer is stopped. While it is best practice to use
  * stateless listeners, you can consume this event to clean up any thread-based resources
@@ -30,6 +32,30 @@ public class ConsumerStoppedEvent extends KafkaEvent {
 
 	private static final long serialVersionUID = 1L;
 
+	public enum Reason {
+
+		/**
+		 * The consumer was stopped because the container was stopped.
+		 */
+		NORMAL,
+
+		/**
+		 * The transactional producer was fenced anf the container
+		 * {@code stopContainerWhenFenced} property is true.
+		 */
+		FENCED,
+
+		/**
+		 * A {@link java.lang.Error} was thrown.
+		 */
+		ERROR
+
+	}
+
+	private final Reason reason;
+
+	private final Object container;
+
 	/**
 	 * Construct an instance with the provided source.
 	 * @param source the container.
@@ -45,13 +71,52 @@ public ConsumerStoppedEvent(Object source) {
 	 * @param container the container or the parent container if the container is a child.
 	 * @since 2.2.1
 	 */
+	@Deprecated
 	public ConsumerStoppedEvent(Object source, Object container) {
+		this(source, container, null, Reason.NORMAL);
+	}
+
+	/**
+	 * Construct an instance with the provided source and container.
+	 * @param source the container instance that generated the event.
+	 * @param container the container or the parent container if the container is a child.
+	 * @param childContainer the child container, or null.
+	 * @param reason the reason.
+	 * @since 2.5.8
+	 */
+	public ConsumerStoppedEvent(Object source, Object container, @Nullable Object childContainer,
+			Reason reason) {
+
 		super(source, container);
+		this.container = childContainer;
+		this.reason = reason;
+	}
+
+	/**
+	 * Return the reason why the consumer was stopped.
+	 * @return the reason.
+	 * @since 2.5.8
+	 */
+	public Reason getReason() {
+		return this.reason;
+	}
+
+	/**
+	 * Return the container that the Consumer belonged to.
+	 * @param <T> the container type.
+	 * @return the container.
+	 * @since 2.5.8
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> T getContainer() {
+		return this.container == null ? (T) getSource() : (T) this.container;
 	}
 
 	@Override
 	public String toString() {
-		return "ConsumerStoppedEvent [source=" + getSource() + "]";
+		return "ConsumerStoppedEvent [source=" + getSource()
+				+ (this.container == null ? "" : (", container=" + this.container))
+				+ ", reason=" + this.reason + "]";
 	}
 
 }

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

@@ -254,6 +254,8 @@ public enum EOSMode {
 
 	private TransactionDefinition transactionDefinition;
 
+	private boolean stopContainerWhenFenced;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -745,6 +747,30 @@ public void setAdviceChain(Advice... adviceChain) {
 		}
 	}
 
+	/**
+	 * When true, the container will stop after a
+	 * {@link org.apache.kafka.common.errors.ProducerFencedException}.
+	 * @return the stopContainerWhenFenced
+	 * @since 2.5.8
+	 */
+	public boolean isStopContainerWhenFenced() {
+		return this.stopContainerWhenFenced;
+	}
+
+	/**
+	 * Set to true to stop the container when a
+	 * {@link org.apache.kafka.common.errors.ProducerFencedException} is thrown.
+	 * Currently, there is no way to determine if such an exception is thrown due to a
+	 * rebalance Vs. a timeout. We therefore cannot call the after rollback processor. The
+	 * best solution is to ensure that the {@code transaction.timeout.ms} is large enough
+	 * so that transactions don't time out.
+	 * @param stopContainerWhenFenced true to stop the container.
+	 * @since 2.5.8
+	 */
+	public void setStopContainerWhenFenced(boolean stopContainerWhenFenced) {
+		this.stopContainerWhenFenced = stopContainerWhenFenced;
+	}
+
 	private void adviseListenerIfNeeded() {
 		if (!CollectionUtils.isEmpty(this.adviceChain)) {
 			if (AopUtils.isAopProxy(this.messageListener)) {

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

@@ -80,6 +80,7 @@
 import org.springframework.kafka.event.ConsumerStartedEvent;
 import org.springframework.kafka.event.ConsumerStartingEvent;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
+import org.springframework.kafka.event.ConsumerStoppedEvent.Reason;
 import org.springframework.kafka.event.ConsumerStoppingEvent;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.event.ListenerContainerNoLongerIdleEvent;
@@ -401,9 +402,15 @@ private void publishConsumerStoppingEvent(Consumer<?, ?> consumer) {
 		}
 	}
 
-	private void publishConsumerStoppedEvent() {
+	private void publishConsumerStoppedEvent(@Nullable Throwable throwable) {
 		if (getApplicationEventPublisher() != null) {
-			getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.thisOrParentContainer));
+			getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.thisOrParentContainer,
+					this.thisOrParentContainer.equals(this) ? null : this,
+					throwable instanceof Error
+						? Reason.ERROR
+						: throwable instanceof StopAfterFenceException
+							? Reason.FENCED
+							: Reason.NORMAL));
 		}
 	}
 
@@ -1076,20 +1083,20 @@ public void run() {
 							+ "' has been fenced");
 					break;
 				}
-				catch (Exception e) {
-					handleConsumerException(e);
-				}
-				catch (Error e) { // NOSONAR - rethrown
+				catch (StopAfterFenceException | Error e) { // NOSONAR - rethrown
 					Runnable runnable = KafkaMessageListenerContainer.this.emergencyStop;
 					if (runnable != null) {
 						runnable.run();
 					}
 					this.logger.error(e, "Stopping container due to an Error");
-					wrapUp();
+					wrapUp(e);
 					throw e;
 				}
+				catch (Exception e) {
+					handleConsumerException(e);
+				}
 			}
-			wrapUp();
+			wrapUp(null);
 		}
 
 		private void setupSeeks() {
@@ -1331,7 +1338,7 @@ private void idleBetweenPollIfNecessary() {
 			}
 		}
 
-		private void wrapUp() {
+		private void wrapUp(@Nullable Throwable throwable) {
 			KafkaUtils.clearConsumerGroupId();
 			if (this.micrometerHolder != null) {
 				this.micrometerHolder.destroy();
@@ -1370,7 +1377,7 @@ private void wrapUp() {
 				this.consumerSeekAwareListener.unregisterSeekCallback();
 			}
 			this.logger.info(() -> getGroupId() + ": Consumer stopped");
-			publishConsumerStoppedEvent();
+			publishConsumerStoppedEvent(throwable);
 		}
 
 		/**
@@ -1540,6 +1547,9 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 				this.logger.error(e, "Producer or '"
 						+ ConsumerConfig.GROUP_INSTANCE_ID_CONFIG
 						+ "' fenced during transaction");
+				if (this.containerProperties.isStopContainerWhenFenced()) {
+					throw new StopAfterFenceException("Container stopping due to fencing", e);
+				}
 			}
 			catch (RuntimeException e) {
 				this.logger.error(e, "Transaction rolled back");
@@ -1806,6 +1816,9 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 				}
 				catch (ProducerFencedException | FencedInstanceIdException e) {
 					this.logger.error(e, "Producer or 'group.instance.id' fenced during transaction");
+					if (this.containerProperties.isStopContainerWhenFenced()) {
+						throw new StopAfterFenceException("Container stopping due to fencing", e);
+					}
 					break;
 				}
 				catch (RuntimeException e) {
@@ -2093,12 +2106,7 @@ else if (this.producer != null
 				this.acks.add(record);
 			}
 			if (this.producer != null) {
-				try {
-					sendOffsetsToTransaction();
-				}
-				catch (Exception e) {
-					this.logger.error(e, "Send offsets to transaction failed");
-				}
+				sendOffsetsToTransaction();
 			}
 		}
 
@@ -2848,4 +2856,13 @@ public void onSuccess(Object result) {
 
 	}
 
+	@SuppressWarnings("serial")
+	private static class StopAfterFenceException extends KafkaException {
+
+		StopAfterFenceException(String message, Throwable t) {
+			super(message, t);
+		}
+
+	}
+
 }

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

@@ -84,6 +84,7 @@
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
+import org.springframework.kafka.event.ConsumerStoppedEvent.Reason;
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.listener.ContainerProperties.AssignmentCommitOption;
@@ -172,15 +173,29 @@ public void testConsumeAndProduceTransactionKTM_BETA() throws Exception {
 		testConsumeAndProduceTransactionGuts(false, false, AckMode.RECORD, EOSMode.BETA);
 	}
 
+	@Test
+	public void testConsumeAndProduceTransactionStopWhenFenced() throws Exception {
+		testConsumeAndProduceTransactionGuts(false, false, AckMode.RECORD, EOSMode.BETA, true);
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handleError, AckMode ackMode,
 			EOSMode eosMode) throws Exception {
 
+		testConsumeAndProduceTransactionGuts(chained, handleError, ackMode, eosMode, false);
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handleError, AckMode ackMode,
+			EOSMode eosMode, boolean stopWhenFenced) throws Exception {
+
 		Consumer consumer = mock(Consumer.class);
+		AtomicBoolean assigned = new AtomicBoolean();
 		final TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
 			((ConsumerRebalanceListener) i.getArgument(1))
 					.onPartitionsAssigned(Collections.singletonList(topicPartition));
+			assigned.set(true);
 			return null;
 		}).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
 		ConsumerRecords records = new ConsumerRecords(Collections.singletonMap(topicPartition,
@@ -199,6 +214,14 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		ConsumerFactory cf = mock(ConsumerFactory.class);
 		willReturn(consumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides());
 		Producer producer = mock(Producer.class);
+		if (stopWhenFenced) {
+			willAnswer(inv -> {
+				if (assigned.get()) {
+					throw new ProducerFencedException("fenced");
+				}
+				return null;
+			}).given(producer).sendOffsetsToTransaction(any(), any(ConsumerGroupMetadata.class));
+		}
 		given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>());
 		final CountDownLatch closeLatch = new CountDownLatch(2);
 		willAnswer(i -> {
@@ -224,6 +247,7 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		props.setTransactionManager(ptm);
 		props.setAssignmentCommitOption(AssignmentCommitOption.ALWAYS);
 		props.setEosMode(eosMode);
+		props.setStopContainerWhenFenced(stopWhenFenced);
 		ConsumerGroupMetadata consumerGroupMetadata = new ConsumerGroupMetadata("group");
 		given(consumer.groupMetadata()).willReturn(consumerGroupMetadata);
 		final KafkaTemplate template = new KafkaTemplate(pf);
@@ -260,6 +284,14 @@ public void onMessage(Object data) {
 		if (handleError) {
 			container.setErrorHandler((e, data) -> { });
 		}
+		CountDownLatch stopEventLatch = new CountDownLatch(1);
+		AtomicReference<ConsumerStoppedEvent> stopEvent = new AtomicReference<>();
+		container.setApplicationEventPublisher(event -> {
+			if (event instanceof ConsumerStoppedEvent) {
+				stopEvent.set((ConsumerStoppedEvent) event);
+				stopEventLatch.countDown();
+			}
+		});
 		container.start();
 		assertThat(closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		InOrder inOrder = inOrder(producer);
@@ -272,27 +304,37 @@ public void onMessage(Object data) {
 			inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
 					new OffsetAndMetadata(0)), consumerGroupMetadata);
 		}
-		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).close(any());
-		inOrder.verify(producer).beginTransaction();
-		ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
-		inOrder.verify(producer).send(captor.capture(), any(Callback.class));
-		assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
-		if (eosMode.equals(EOSMode.ALPHA)) {
-			inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
-					new OffsetAndMetadata(1)), "group");
+		if (stopWhenFenced) {
+			assertThat(stopEventLatch.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(stopEvent.get().getReason()).isEqualTo(Reason.FENCED);
 		}
 		else {
-			inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
-					new OffsetAndMetadata(1)), consumerGroupMetadata);
+			inOrder.verify(producer).commitTransaction();
+			inOrder.verify(producer).close(any());
+			inOrder.verify(producer).beginTransaction();
+			ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
+			inOrder.verify(producer).send(captor.capture(), any(Callback.class));
+			assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
+			if (eosMode.equals(EOSMode.ALPHA)) {
+				inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
+						new OffsetAndMetadata(1)), "group");
+			}
+			else {
+				inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
+						new OffsetAndMetadata(1)), consumerGroupMetadata);
+			}
+			inOrder.verify(producer).commitTransaction();
+			inOrder.verify(producer).close(any());
+			container.stop();
+			verify(pf, times(2)).createProducer(isNull());
+			verifyNoMoreInteractions(producer);
+			assertThat(transactionalIds.get(0)).isEqualTo("group.foo.0");
+			assertThat(transactionalIds.get(0)).isEqualTo("group.foo.0");
+			assertThat(stopEventLatch.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(stopEvent.get().getReason()).isEqualTo(Reason.NORMAL);
 		}
-		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).close(any());
-		container.stop();
-		verify(pf, times(2)).createProducer(isNull());
-		verifyNoMoreInteractions(producer);
-		assertThat(transactionalIds.get(0)).isEqualTo("group.foo.0");
-		assertThat(transactionalIds.get(0)).isEqualTo("group.foo.0");
+		MessageListenerContainer stoppedContainer = stopEvent.get().getContainer();
+		assertThat(stoppedContainer).isSameAs(container);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })