GH-2350: Fix Paused Partitions After Rebalance

Resolves #2350

Do not maintain the `pausePartitionsRequested` field in the concurrent MLC,
it is not used and can cause confusion.

Also fixes some synchronization around `CMLC.containers`.

Also resolves #2222

The previous fix removed revoked partitions from `pausePartitionsRequested`;
this was incorrect - consider a rebalance where we lose `topic-0` and another
rebalance where we are re-assigned `topic-0`. According to the contract, this
partition should remain paused.

**cherry-pick to 2.9.x, 2.8.x**

Author: garyrussell

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

@@ -347,7 +347,6 @@ public void resume() {
 	@Override
 	public void pausePartition(TopicPartition topicPartition) {
 		synchronized (this.lifecycleMonitor) {
-			super.pausePartition(topicPartition);
 			this.containers
 					.stream()
 					.filter(container -> containsPartition(topicPartition, container))
@@ -358,7 +357,6 @@ public void pausePartition(TopicPartition topicPartition) {
 	@Override
 	public void resumePartition(TopicPartition topicPartition) {
 		synchronized (this.lifecycleMonitor) {
-			super.resumePartition(topicPartition);
 			this.containers
 					.stream()
 					.filter(container -> containsPartition(topicPartition, container))
@@ -368,18 +366,22 @@ public void resumePartition(TopicPartition topicPartition) {
 
 	@Override
 	public boolean isPartitionPaused(TopicPartition topicPartition) {
-		return this
-				.containers
-				.stream()
-				.anyMatch(container -> container.isPartitionPaused(topicPartition));
+		synchronized (this.lifecycleMonitor) {
+			return this
+					.containers
+					.stream()
+					.anyMatch(container -> container.isPartitionPaused(topicPartition));
+		}
 	}
 
 	@Override
 	public boolean isInExpectedState() {
-		return (isRunning() || isStoppedNormally()) && this.containers
-				.stream()
-				.map(container -> container.isInExpectedState())
-				.allMatch(bool -> Boolean.TRUE.equals(bool));
+		synchronized (this.lifecycleMonitor) {
+			return (isRunning() || isStoppedNormally()) && this.containers
+					.stream()
+					.map(container -> container.isInExpectedState())
+					.allMatch(bool -> Boolean.TRUE.equals(bool));
+		}
 	}
 
 	private boolean containsPartition(TopicPartition topicPartition, KafkaMessageListenerContainer<K, V> container) {

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

@@ -3491,7 +3491,6 @@ private void repauseIfNeeded(Collection<TopicPartition> partitions) {
 					publishConsumerPausedEvent(toRepause);
 				}
 				this.revoked.removeAll(toRepause);
-				this.revoked.forEach(tp -> resumePartition(tp));
 				ListenerConsumer.this.pausedPartitions.removeAll(this.revoked);
 				this.revoked.clear();
 				if (ListenerConsumer.this.pausedForNack.size() > 0) {

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -2803,6 +2803,11 @@ public void rePausePartitionAfterRebalance() throws Exception {
 				.asInstanceOf(InstanceOfAssertFactories.collection(TopicPartition.class))
 				.hasSize(1)
 				.contains(tp0);
+		assertThat(container)
+				.extracting("pauseRequestedPartitions")
+				.asInstanceOf(InstanceOfAssertFactories.collection(TopicPartition.class))
+				.hasSize(2)
+				.contains(tp0, tp1);
 		suspendConsumerThread.countDown();
 		container.stop();
 	}