Improve Kafka Exception Message

- when we can't determine reply topic

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

@@ -626,7 +626,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 				replyTopic = getSingleReplyTopic();
 			}
 			else {
-				throw new IllegalStateException("No reply topic header and no default reply topic can be determined");
+				throw new IllegalStateException("No reply topic header and no default reply topic can be determined; "
+						+ "container's assigned partitions: " + this.replyTopicsAndPartitions);
 			}
 		}
 		else {

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java

@@ -459,7 +459,9 @@ public class KafkaDslTests {
 
 				@Override
 				public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
-					ContextConfiguration.this.replyContainerLatch.countDown();
+					if (!partitions.isEmpty()) {
+						ContextConfiguration.this.replyContainerLatch.countDown();
+					}
 				}
 
 			});

spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt

@@ -345,7 +345,9 @@ class KafkaDslKotlinTests {
 				}
 
 				override fun onPartitionsAssigned(partitions: Collection<TopicPartition>) {
-					this@ContextConfiguration.replyContainerLatch.countDown()
+					if (!partitions.isEmpty()) {
+						this@ContextConfiguration.replyContainerLatch.countDown()
+					}
 				}
 
 			})