GH-2178: Fix CSA.onPartitionsAssigned (Manual)

Resolves #2178

The wrong collection was used as the source for the `assignments` map,

`SeekPosition.BEGINNING` and `.END` entries are removed; use the
`definedPartitions` field instead.

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

Author: garyrussell

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

@@ -2885,7 +2885,7 @@ private void initPartitionsIfNeeded() {
 			});
 			doInitialSeeks(partitions, beginnings, ends);
 			if (this.consumerSeekAwareListener != null) {
-				this.consumerSeekAwareListener.onPartitionsAssigned(partitions.keySet().stream()
+				this.consumerSeekAwareListener.onPartitionsAssigned(this.definedPartitions.keySet().stream()
 							.map(tp -> new SimpleEntry<>(tp, this.consumer.position(tp)))
 							.collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue())),
 						this.seekCallback);

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

@@ -2790,7 +2790,21 @@ public void testInitialSeek() throws Exception {
 		containerProps.setGroupId("grp");
 		containerProps.setAckMode(AckMode.RECORD);
 		containerProps.setClientId("clientId");
-		containerProps.setMessageListener((MessageListener) r -> { });
+
+		Map<TopicPartition, Long> assigned = new HashMap<>();
+		class Listener extends AbstractConsumerSeekAware implements MessageListener {
+
+			@Override
+			public void onMessage(Object data) {
+			}
+
+			@Override
+			public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
+				assigned.putAll(assignments);
+			}
+
+		}
+		containerProps.setMessageListener(new Listener());
 		containerProps.setMissingTopicsFatal(false);
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
@@ -2808,6 +2822,7 @@ public void testInitialSeek() throws Exception {
 		verify(consumer).seek(new TopicPartition("foo", 3), Long.MAX_VALUE);
 		verify(consumer).seek(new TopicPartition("foo", 6), 42L);
 		container.stop();
+		assertThat(assigned).hasSize(8);
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })