GH-1446: Support Static Group Membership

garyrussell · garyrussell · commit 978868ecd299 · 2020-04-15T11:58:58.000-04:00
Resolves #1446 See https://kafka.apache.org/documentation/#static_membership Add a qualifier to `group.instance.id` for each child container. Catch `FencedInstanceIdException` and stop the container. Fix delay on self-stopping containers. Treat `FencedInstanceIdException` the same as `ProducerFencedException` when using transactions (do not call `AfterRollbackProcessor`. **cherry-pick to 2.4.x, 2.3.x** * Add docs. * Fix container stopped event publishing. * Publish stop event after actually stopped. * Restore publishContainerStoppedEvent() to proper place.
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
@@ -406,16 +406,32 @@ public void checkGroupId() {
 
 	@Override
 	public final void stop() {
+		stop(true);
+	}
+
+	/**
+	 * Stop the container.
+	 * @param wait wait for the listener to terminate.
+	 * @since 2.3.8
+	 */
+	public final void stop(boolean wait) {
 		synchronized (this.lifecycleMonitor) {
 			if (isRunning()) {
-				final CountDownLatch latch = new CountDownLatch(1);
-				doStop(latch::countDown);
-				try {
-					latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR
-					publishContainerStoppedEvent();
+				if (wait) {
+					final CountDownLatch latch = new CountDownLatch(1);
+					doStop(latch::countDown);
+					try {
+						latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR
+						publishContainerStoppedEvent();
+					}
+					catch (@SuppressWarnings("unused") InterruptedException e) {
+						Thread.currentThread().interrupt();
+					}
 				}
-				catch (@SuppressWarnings("unused") InterruptedException e) {
-					Thread.currentThread().interrupt();
+				else {
+					doStop(() -> {
+						publishContainerStoppedEvent();
+					});
 				}
 			}
 		}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
@@ -54,6 +54,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.errors.WakeupException;
 
@@ -97,6 +98,7 @@
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.StringUtils;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.ListenableFutureCallback;
 
@@ -611,6 +613,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 		@SuppressWarnings(UNCHECKED)
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
 			Properties consumerProperties = new Properties(this.containerProperties.getKafkaConsumerProperties());
+			checkGroupInstance(consumerProperties, KafkaMessageListenerContainer.this.consumerFactory);
 			this.autoCommit = determineAutoCommit(consumerProperties);
 			this.consumer =
 					KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
@@ -689,6 +692,23 @@ else if (listener instanceof MessageListener) {
 			this.micrometerHolder = obtainMicrometerHolder();
 		}
 
+		private void checkGroupInstance(Properties properties, ConsumerFactory<K, V> consumerFactory) {
+			String groupInstance = properties.getProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG);
+			if (!StringUtils.hasText(groupInstance)) {
+				Object factoryConfig = consumerFactory.getConfigurationProperties()
+						.get(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG);
+				if (factoryConfig instanceof String) {
+					groupInstance = (String) factoryConfig;
+				}
+			}
+			if (StringUtils.hasText(KafkaMessageListenerContainer.this.clientIdSuffix)
+					&& StringUtils.hasText(groupInstance)) {
+
+				properties.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG,
+						groupInstance + KafkaMessageListenerContainer.this.clientIdSuffix);
+			}
+		}
+
 		private boolean determineCommitCurrent(Properties consumerProperties) {
 			if (AssignmentCommitOption.NEVER.equals(this.autoCommitOption)) {
 				return false;
@@ -951,6 +971,12 @@ public void run() {
 						sleepFor(this.authorizationExceptionRetryInterval);
 					}
 				}
+				catch (FencedInstanceIdException fie) {
+					this.fatalError = true;
+					ListenerConsumer.this.logger.error(fie, "'" + ConsumerConfig.GROUP_INSTANCE_ID_CONFIG
+							+ "' has been fenced");
+					break;
+				}
 				catch (Exception e) {
 					handleConsumerException(e);
 				}
@@ -1144,7 +1170,7 @@ private void wrapUp() {
 			}
 			else {
 				this.logger.error("Fatal consumer exception; stopping container");
-				KafkaMessageListenerContainer.this.stop();
+				KafkaMessageListenerContainer.this.stop(false);
 			}
 			this.monitorTask.cancel(true);
 			if (!this.taskSchedulerExplicitlySet) {
@@ -1305,8 +1331,10 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 					}
 				});
 			}
-			catch (ProducerFencedException e) {
-				this.logger.error(e, "Producer fenced during transaction");
+			catch (ProducerFencedException | FencedInstanceIdException e) {
+				this.logger.error(e, "Producer or '"
+						+ ConsumerConfig.GROUP_INSTANCE_ID_CONFIG
+						+ "' fenced during transaction");
 			}
 			catch (RuntimeException e) {
 				this.logger.error(e, "Transaction rolled back");
@@ -1557,8 +1585,8 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 
 					});
 				}
-				catch (ProducerFencedException e) {
-					this.logger.error(e, "Producer fenced during transaction");
+				catch (ProducerFencedException | FencedInstanceIdException e) {
+					this.logger.error(e, "Producer or 'group.instance.id' fenced during transaction");
 					break;
 				}
 				catch (RuntimeException e) {
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
@@ -68,6 +68,8 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.FencedInstanceIdException;
+import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.serialization.IntegerDeserializer;
 import org.junit.jupiter.api.BeforeAll;
@@ -2676,16 +2678,72 @@ void testFatalErrorOnAuthorizationException() throws Exception {
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 
-		CountDownLatch stopping = new CountDownLatch(1);
+		CountDownLatch stopped = new CountDownLatch(1);
 
 		container.setApplicationEventPublisher(e -> {
-			if (e instanceof ConsumerStoppingEvent) {
-				stopping.countDown();
+			if (e instanceof ConsumerStoppedEvent) {
+				stopped.countDown();
+			}
+		});
+
+		container.start();
+		assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void testNotFatalErrorOnAuthorizationException() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
+		given(cf.getConfigurationProperties()).willReturn(new HashMap<>());
+		CountDownLatch latch = new CountDownLatch(2);
+		willAnswer(invoc -> {
+			latch.countDown();
+			throw new TopicAuthorizationException("test");
+		}).given(consumer).poll(any());
+
+		ContainerProperties containerProps = new ContainerProperties(topic1);
+		containerProps.setGroupId("grp");
+		containerProps.setClientId("clientId");
+		containerProps.setMessageListener((MessageListener) r -> { });
+		containerProps.setAuthorizationExceptionRetryInterval(Duration.ofMillis(100));
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void testFatalErrorOnFencedInstanceException() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
+		given(cf.getConfigurationProperties()).willReturn(new HashMap<>());
+
+		willThrow(FencedInstanceIdException.class)
+				.given(consumer).poll(any());
+
+		ContainerProperties containerProps = new ContainerProperties(topic1);
+		containerProps.setGroupId("grp");
+		containerProps.setClientId("clientId");
+		containerProps.setMessageListener((MessageListener) r -> { });
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+
+		CountDownLatch stopped = new CountDownLatch(1);
+
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof ConsumerStoppedEvent) {
+				stopped.countDown();
 			}
 		});
 
 		container.start();
-		assertThat(stopping.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
 		container.stop();
 	}
 
diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc
@@ -666,6 +666,10 @@ Starting with version 2.3.4, you can set the listener container's `interceptBefo
 
 No interceptor is provided for batch listeners because Kafka already provides a `ConsumerInterceptor`.
 
+Starting with versions 2.3.8, 2.4.6, the `ConcurrentMessageListenerContainer` now supports https://kafka.apache.org/documentation/#static_membership[Static Membership] when the concurrency is greater than one.
+The `group.instance.id` is suffixed with `-n` with `n` starting at `1`.
+This, together with an increased `session.timeout.ms`, can be used to reduce rebalance events, for example, when application instances are restarted.
+
 [[kafka-container]]
 ====== Using `KafkaMessageListenerContainer`
 
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
@@ -78,6 +78,9 @@ See <<container-factory>> for more information
 The `ContainerProperties` provides an `authorizationExceptionRetryInterval` option to let the listener container to retry after any `AuthorizationException` is thrown by the `KafkaConsumer`.
 See its JavaDocs and <<kafka-container>> for more information.
 
+Static group membership is now supported.
+See <<message-listener-container>> for more information.
+
 ==== @KafkaListener
 
 The `@KafkaListener` annotation has a new property `splitIterables`; default true.
