Forbid fenced Container to stop ConcurrentContainer

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

@@ -94,7 +94,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	protected final ReentrantLock lifecycleLock = new ReentrantLock(); // NOSONAR
 
-	protected final AtomicBoolean enforceRebalanceRequested = new AtomicBoolean();
+	private final AtomicBoolean enforceRebalanceRequested = new AtomicBoolean();
 
 	private final Set<TopicPartition> pauseRequestedPartitions = ConcurrentHashMap.newKeySet();
 
@@ -193,6 +193,14 @@ protected AbstractMessageListenerContainer(ConsumerFactory<? super K, ? super V>
 		}
 	}
 
+	/**
+	 *	To be used only with {@link ConcurrentMessageListenerContainerRef}.
+	 */
+	AbstractMessageListenerContainer() {
+		this.containerProperties = null;
+		this.consumerFactory = null;
+	}
+
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
@@ -278,15 +286,27 @@ public boolean isRunning() {
 		return this.running;
 	}
 
-	protected void setFenced(boolean fenced) {
+	void setFenced(boolean fenced) {
 		this.fenced = fenced;
 	}
 
+	protected boolean isFenced() {
+		return this.fenced;
+	}
+
 	@Deprecated(since = "3.2", forRemoval = true)
 	protected boolean isPaused() {
 		return this.paused;
 	}
 
+	protected boolean isEnforceRebalanceRequested() {
+		return this.enforceRebalanceRequested.get();
+	}
+
+	protected void setEnforceRebalanceRequested(boolean enforceRebalance) {
+		this.enforceRebalanceRequested.set(enforceRebalance);
+	}
+
 	@Override
 	public boolean isPartitionPauseRequested(TopicPartition topicPartition) {
 		return this.pauseRequestedPartitions.contains(topicPartition);

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

@@ -305,13 +305,17 @@ private KafkaMessageListenerContainer<K, V> constructContainer(ContainerProperti
 			@Nullable TopicPartitionOffset[] topicPartitions, int i) {
 
 		KafkaMessageListenerContainer<K, V> container;
+		ConcurrentMessageListenerContainerRef concurrentMessageListenerContainerRef =
+				new ConcurrentMessageListenerContainerRef<>(this, this.lifecycleLock);
 		if (topicPartitions == null) {
-			container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, containerProperties); // NOSONAR
+			container = new KafkaMessageListenerContainer<>(concurrentMessageListenerContainerRef, this,
+					this.consumerFactory, containerProperties); // NOSONAR
 		}
 		else {
-			container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, // NOSONAR
-					containerProperties, partitionSubset(containerProperties, i));
+			container = new KafkaMessageListenerContainer<>(concurrentMessageListenerContainerRef, this,
+					this.consumerFactory, containerProperties, partitionSubset(containerProperties, i)); // NOSONAR
 		}
+		concurrentMessageListenerContainerRef.setKafkaMessageListenerContainer(container);
 		return container;
 	}