GH-2565 Fix Resume Partition When Unassigned

Resolves #2565

Previously, when a partition was resumed while not currently assigned,
it would be paused indefinitely when reassigned.

This was caused by the `ConcurrentMessageListenerContainer` only resuming
partitions in child containers if that partition is currently assigned
to the container, leaving the pause request in place.

Use the `isPartitionPauseRequested()` method on the child container instead
when deciding whether to resume the partition.

**cherry-pick to 2.9.x**

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2023 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.
@@ -374,7 +374,7 @@ public void resumePartition(TopicPartition topicPartition) {
 		synchronized (this.lifecycleMonitor) {
 			this.containers
 					.stream()
-					.filter(container -> containsPartition(topicPartition, container))
+					.filter(container -> container.isPartitionPauseRequested(topicPartition))
 					.forEach(container -> container.resumePartition(topicPartition));
 		}
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2022 the original author or authors.
+ * Copyright 2019-2023 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.
@@ -24,6 +24,7 @@
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
@@ -844,6 +845,60 @@ private void testInitialCommitIBasedOnCommitted(boolean committed) throws Interr
 		}
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void removeFromPartitionPauseRequestedWhenNotAssigned() throws InterruptedException {
+		Consumer consumer = mock(Consumer.class);
+		CountDownLatch pollLatch = new CountDownLatch(1);
+		willAnswer(inv -> {
+			pollLatch.countDown();
+			Thread.sleep(50);
+			return ConsumerRecords.empty();
+		}).given(consumer).poll(any());
+		CountDownLatch pauseLatch = new CountDownLatch(1);
+		willAnswer(inv -> {
+			pauseLatch.countDown();
+			return null;
+		}).given(consumer).pause(any());
+		TopicPartition tp0 = new TopicPartition("foo", 0);
+		List<TopicPartition> assignments = Arrays.asList(tp0);
+		AtomicReference<ConsumerRebalanceListener> rebal = new AtomicReference<>();
+		willAnswer(invocation -> {
+			rebal.set(invocation.getArgument(1));
+			rebal.get().onPartitionsAssigned(assignments);
+			return null;
+		}).given(consumer).subscribe(any(Collection.class), any());
+		ConsumerFactory cf = mock(ConsumerFactory.class);
+		given(cf.createConsumer(any(), any(), any(), any())).willReturn(consumer);
+		given(cf.getConfigurationProperties())
+				.willReturn(Collections.singletonMap(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"));
+		ContainerProperties containerProperties = new ContainerProperties("foo");
+		containerProperties.setGroupId("grp");
+		containerProperties.setMessageListener((MessageListener) rec -> { });
+		containerProperties.setMissingTopicsFatal(false);
+		containerProperties.setAssignmentCommitOption(AssignmentCommitOption.LATEST_ONLY);
+		ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer(cf,
+				containerProperties);
+		container.start();
+		assertThat(pollLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.pausePartition(tp0);
+		KafkaMessageListenerContainer child = (KafkaMessageListenerContainer) KafkaTestUtils
+				.getPropertyValue(container, "containers", List.class).get(0);
+		assertThat(child.isPartitionPauseRequested(tp0)).isTrue();
+		assertThat(pauseLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		rebal.get().onPartitionsRevoked(assignments);
+		assertThat(child.isPartitionPauseRequested(tp0)).isTrue();
+		// immediate pause when re-assigned
+		rebal.get().onPartitionsAssigned(assignments);
+		verify(consumer, times(2)).pause(any());
+		rebal.get().onPartitionsRevoked(assignments);
+		// resume partition while unassigned
+		container.resumePartition(tp0);
+		assertThat(child.isPartitionPauseRequested(tp0)).isFalse();
+		rebal.get().onPartitionsAssigned(assignments);
+		verify(consumer, times(2)).pause(any()); // no immediate pause this time
+	}
+
 	public static class TestMessageListener1 implements MessageListener<String, String>, ConsumerSeekAware {
 
 		private static ThreadLocal<ConsumerSeekCallback> callbacks = new ThreadLocal<>();