spring-projectsGH-1688: ConsumerAwareRebalListener Improvements

Resolves spring-projects#1668

Users might implement one of the not-consumer-aware methods so the default
methods should delegate to those.

Call in try/catch block, expecially `...RevokedBeforeCommit` to prevent the
commits to be skipped.

**cherry-pick to 2.6.x**

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -18,10 +18,13 @@
 
 import java.util.Collection;
 
+import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.core.log.LogAccessor;
+
 /**
  * A rebalance listener that provides access to the consumer object. Starting with version
  * 2.1.5, as a convenience, default no-op implementations are provided for all methods,
@@ -33,14 +36,24 @@
  */
 public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener {
 
+	/**
+	 * {@link LogAccessor} for use in default methods.
+	 */
+	LogAccessor logger = new LogAccessor(LogFactory.getLog(ConsumerAwareRebalanceListener.class));
+
 	/**
 	 * The same as {@link #onPartitionsRevoked(Collection)} with the additional consumer
 	 * parameter. It is invoked by the container before any pending offsets are committed.
 	 * @param consumer the consumer.
 	 * @param partitions the partitions.
 	 */
 	default void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
-		// do nothing
+		try {
+			onPartitionsRevoked(partitions);
+		}
+		catch (Exception e) { // NOSONAR
+			logger.debug(e, "User method threw exception");
+		}
 	}
 
 	/**
@@ -50,7 +63,6 @@ default void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection
 	 * @param partitions the partitions.
 	 */
 	default void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
-		// do nothing
 	}
 
 	/**
@@ -60,7 +72,12 @@ default void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<
 	 * @since 2.4
 	 */
 	default void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
-		// do nothing
+		try {
+			onPartitionsLost(partitions);
+		}
+		catch (Exception e) { // NOSONAR
+			logger.debug(e, "User method threw exception");
+		}
 	}
 
 	/**
@@ -70,22 +87,24 @@ default void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition
 	 * @param partitions the partitions.
 	 */
 	default void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
-		// do nothing
+		try {
+			onPartitionsAssigned(partitions);
+		}
+		catch (Exception e) { // NOSONAR
+			logger.debug(e, "User method threw exception");
+		}
 	}
 
 	@Override
 	default void onPartitionsRevoked(Collection<TopicPartition> partitions) {
-		throw new UnsupportedOperationException("Listener container should never call this");
 	}
 
 	@Override
 	default void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-		throw new UnsupportedOperationException("Listener container should never call this");
 	}
 
 	@Override
 	default void onPartitionsLost(Collection<TopicPartition> partitions) {
-		throw new UnsupportedOperationException("Listener container should never call this");
 	}
 
 }

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

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Gary Russell
+ * @since 2.6.5
+ *
+ */
+public class ConsumerAwareRebalanceListenerTests {
+
+	@Test
+	void nonConsumerAwareTestAssigned() {
+		AtomicBoolean called = new AtomicBoolean();
+		new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+				called.set(true);
+			}
+
+		}.onPartitionsAssigned(null, null);
+		assertThat(called.get()).isTrue();
+	}
+
+	@Test
+	void nonConsumerAwareTestAssignedThrows() {
+		AtomicBoolean called = new AtomicBoolean();
+		new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+				called.set(true);
+				throw new RuntimeException();
+			}
+
+		}.onPartitionsAssigned(null, null);
+		assertThat(called.get()).isTrue();
+	}
+
+	@Test
+	void nonConsumerAwareTestRevoked() {
+		AtomicBoolean called = new AtomicBoolean();
+		new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+				called.set(true);
+			}
+
+		}.onPartitionsRevokedBeforeCommit(null, null);
+		assertThat(called.get()).isTrue();
+	}
+
+	@Test
+	void nonConsumerAwareTestRevokedThrows() {
+		AtomicBoolean called = new AtomicBoolean();
+		new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+				called.set(true);
+				throw new RuntimeException();
+			}
+
+		}.onPartitionsRevokedBeforeCommit(null, null);
+		assertThat(called.get()).isTrue();
+	}
+
+	@Test
+	void nonConsumerAwareTestLost() {
+		AtomicBoolean called = new AtomicBoolean();
+		new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsLost(Collection<TopicPartition> partitions) {
+				called.set(true);
+			}
+
+		}.onPartitionsLost(null, null);
+		assertThat(called.get()).isTrue();
+	}
+
+	@Test
+	void nonConsumerAwareTestLostThrows() {
+		AtomicBoolean called = new AtomicBoolean();
+		new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsLost(Collection<TopicPartition> partitions) {
+				called.set(true);
+				throw new RuntimeException();
+			}
+
+		}.onPartitionsLost(null, null);
+		assertThat(called.get()).isTrue();
+	}
+
+}