Replying template - add getter for reply topic(s)

- supports the SIK outbound gateway to determine if there's a default reply topic.
- supports the SIK validating a reply topic header if present in the outbound message.

* Use `assignedPartitions` instead

* Polishing

Author: garyrussell

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

@@ -196,6 +196,7 @@ public int getPhase() {
 		return this.phase;
 	}
 
+	@Override
 	public ContainerProperties getContainerProperties() {
 		return this.containerProperties;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -18,6 +18,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -26,6 +27,7 @@
 
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.config.ContainerProperties;
@@ -95,6 +97,18 @@ public List<KafkaMessageListenerContainer<K, V>> getContainers() {
 		return Collections.unmodifiableList(this.containers);
 	}
 
+	@Override
+	public Collection<TopicPartition> getAssignedPartitions() {
+		List<TopicPartition> assigned = new ArrayList<>();
+		this.containers.forEach(c -> {
+			Collection<TopicPartition> assignedPartitions = c.getAssignedPartitions();
+			if (assignedPartitions != null) {
+				assigned.addAll(assignedPartitions);
+			}
+		});
+		return assigned;
+	}
+
 	@Override
 	public Map<String, Map<MetricName, ? extends Metric>> metrics() {
 		Map<String, Map<MetricName, ? extends Metric>> metrics = new HashMap<>();

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

@@ -180,6 +180,7 @@ public void setClientIdSuffix(String clientIdSuffix) {
 	 * @return the {@link TopicPartition}s currently assigned to this container,
 	 * either explicitly or by Kafka; may be null if not assigned yet.
 	 */
+	@Override
 	public Collection<TopicPartition> getAssignedPartitions() {
 		ListenerConsumer listenerConsumer = this.listenerConsumer;
 		if (listenerConsumer != null) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -16,12 +16,15 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.Collection;
 import java.util.Map;
 
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.context.SmartLifecycle;
+import org.springframework.kafka.listener.config.ContainerProperties;
 
 /**
  * Internal abstraction used by the framework representing a message
@@ -48,4 +51,22 @@ public interface MessageListenerContainer extends SmartLifecycle {
 	 */
 	Map<String, Map<MetricName, ? extends Metric>> metrics();
 
+	/**
+	 * Return the container properties for this container.
+	 * @return the properties.
+	 * @since 2.1.3
+	 */
+	default ContainerProperties getContainerProperties() {
+		throw new UnsupportedOperationException("This container doesn't support retrieving its properties");
+	}
+
+	/**
+	 * Return the assigned topics/partitions for this container.
+	 * @return the topics/partitions.
+	 * @since 2.1.3
+	 */
+	default Collection<TopicPartition> getAssignedPartitions() {
+		throw new UnsupportedOperationException("This container doesn't support retrieving its assigned partitions");
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

@@ -18,6 +18,7 @@
 
 import java.nio.ByteBuffer;
 import java.time.Instant;
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.UUID;
@@ -26,6 +27,7 @@
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.internals.RecordHeader;
 
@@ -100,6 +102,37 @@ public void setReplyTimeout(long replyTimeout) {
 		this.replyTimeout = replyTimeout;
 	}
 
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	/**
+	 * Return the topics/partitions assigned to the replying listener container.
+	 * @return the topics/partitions.
+	 */
+	public Collection<TopicPartition> getAssignedReplyTopicPartitions() {
+		return this.replyContainer.getAssignedPartitions();
+	}
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		if (!this.schedulerSet) {
@@ -130,29 +163,6 @@ public synchronized void stop() {
 		}
 	}
 
-	@Override
-	public boolean isRunning() {
-		return this.running;
-	}
-
-	@Override
-	public int getPhase() {
-		return this.phase;
-	}
-
-	public void setPhase(int phase) {
-		this.phase = phase;
-	}
-
-	@Override
-	public boolean isAutoStartup() {
-		return this.autoStartup;
-	}
-
-	public void setAutoStartup(boolean autoStartup) {
-		this.autoStartup = autoStartup;
-	}
-
 	@Override
 	public void stop(Runnable callback) {
 		stop();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -119,6 +119,7 @@ public void testAutoCommit() throws Exception {
 		container.start();
 
 		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+		assertThat(container.getAssignedPartitions().size()).isEqualTo(2);
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);

spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java

@@ -190,6 +190,8 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		ReplyingKafkaTemplate<Integer, String, String> template = new ReplyingKafkaTemplate<>(this.config.pf(), container);
 		template.start();
 		assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
+		assertThat(template.getAssignedReplyTopicPartitions().size()).isEqualTo(5);
+		assertThat(template.getAssignedReplyTopicPartitions().iterator().next().topic()).isEqualTo(topic);
 		return template;
 	}