GH-1974: Fix Default Task Executor

Resolves #1974

A new `SimpleAsyncTaskExecutor` was created for each child container when
the container was stopped/started.
Reuse the same container for each child; handle an increase to `concurrency` between
stop and start.

* Fix unnecessary bean name null check.

* Fix previous commit.

Author: garyrussell

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

@@ -33,6 +33,8 @@
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.task.AsyncListenableTaskExecutor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.lang.Nullable;
@@ -60,6 +62,8 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 
 	private final List<KafkaMessageListenerContainer<K, V>> containers = new ArrayList<>();
 
+	private final List<AsyncListenableTaskExecutor> executors = new ArrayList<>();
+
 	private int concurrency = 1;
 
 	private boolean alwaysClientIdSuffix = true;
@@ -210,7 +214,8 @@ protected void doStart() {
 
 	private void configureChildContainer(int index, KafkaMessageListenerContainer<K, V> container) {
 		String beanName = getBeanName();
-		container.setBeanName((beanName != null ? beanName : "consumer") + "-" + index);
+		beanName = (beanName == null ? "consumer" : beanName) + "-" + index;
+		container.setBeanName(beanName);
 		ApplicationContext applicationContext = getApplicationContext();
 		if (applicationContext != null) {
 			container.setApplicationContext(applicationContext);
@@ -230,6 +235,17 @@ private void configureChildContainer(int index, KafkaMessageListenerContainer<K,
 			stopAbnormally(() -> {
 			});
 		});
+		AsyncListenableTaskExecutor exec = container.getContainerProperties().getConsumerTaskExecutor();
+		if (exec == null) {
+			if ((this.executors.size() > index)) {
+				exec = this.executors.get(index);
+			}
+			else {
+				exec = new SimpleAsyncTaskExecutor(beanName + "-C-");
+				this.executors.add(exec);
+			}
+			container.getContainerProperties().setConsumerTaskExecutor(exec);
+		}
 	}
 
 	private KafkaMessageListenerContainer<K, V> constructContainer(ContainerProperties containerProperties,

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

@@ -79,7 +79,8 @@
 		ConcurrentMessageListenerContainerTests.topic6, ConcurrentMessageListenerContainerTests.topic7,
 		ConcurrentMessageListenerContainerTests.topic8, ConcurrentMessageListenerContainerTests.topic9,
 		ConcurrentMessageListenerContainerTests.topic10, ConcurrentMessageListenerContainerTests.topic11,
-		ConcurrentMessageListenerContainerTests.topic12 })
+		ConcurrentMessageListenerContainerTests.topic12 },
+			brokerProperties = "group.initial.rebalance.delay.ms:500")
 public class ConcurrentMessageListenerContainerTests {
 
 	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
@@ -321,6 +322,19 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 			latch.countDown();
 		});
 
+		Set<String> listenerThreadNames = new ConcurrentSkipListSet<>();
+		CountDownLatch rebalLatch = new CountDownLatch(5);
+		containerProps.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+				if (listenerThreadNames.add(Thread.currentThread().getName())) {
+					rebalLatch.countDown();
+				}
+			}
+
+		});
+
 		ConcurrentMessageListenerContainer<Integer, String> container =
 				new ConcurrentMessageListenerContainer<>(cf, containerProps);
 		container.setConcurrency(2);
@@ -340,8 +354,15 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 		template.flush();
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		container.stop();
-		this.logger.info("Stop manual");
+		container.setConcurrency(3);
+		container.start();
+		assertThat(rebalLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		assertThat(listenerThreadNames)
+				.extracting(str -> str.substring(str.length() - 5))
+				.containsExactlyInAnyOrder("0-C-1", "1-C-1", "0-C-2", "1-C-2", "2-C-1");
 		assertThat(overrides.get().getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isEqualTo("false");
+		this.logger.info("Stop manual");
 	}
 
 	@Test