GH-2332: Fix Partitions in Pause Event

Author: garyrussell

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

@@ -1700,7 +1700,7 @@ private void doPauseConsumerIfNecessary() {
 					this.consumerPaused = true;
 					this.pauseForPending = false;
 					this.logger.debug(() -> "Paused consumption from: " + this.consumer.paused());
-					publishConsumerPausedEvent(this.consumer.assignment());
+					publishConsumerPausedEvent(assigned);
 				}
 			}
 		}

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

@@ -56,11 +56,13 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.test.annotation.DirtiesContext;
@@ -114,6 +116,9 @@ public void pausesWithManualAssignment() throws Exception {
 		assertThat(this.config.count).isEqualTo(4);
 		assertThat(this.config.contents).contains("foo", "bar", "baz", "qux");
 		verify(this.consumer, never()).seek(any(), anyLong());
+		assertThat(this.config.eventLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.event.getPartitions()).contains(
+				new TopicPartition("foo", 0), new TopicPartition("foo", 1), new TopicPartition("foo", 2));
 	}
 
 	@Configuration
@@ -130,8 +135,12 @@ public static class Config {
 
 		final CountDownLatch commitLatch = new CountDownLatch(3);
 
+		final CountDownLatch eventLatch = new CountDownLatch(1);
+
 		int count;
 
+		volatile ConsumerPausedEvent event;
+
 		@KafkaListener(id = "id", groupId = "grp",
 				topicPartitions = @org.springframework.kafka.annotation.TopicPartition(topic = "foo",
 						partitions = "#{'0,1,2'.split(',')}"))
@@ -228,6 +237,12 @@ ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(KafkaListe
 			return factory;
 		}
 
+		@EventListener
+		public void paused(ConsumerPausedEvent event) {
+			this.event = event;
+			this.eventLatch.countDown();
+		}
+
 	}
 
 }