diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java index 66c5b33d2e..08700f6f85 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaProducerContext.java @@ -27,6 +27,7 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.context.SmartLifecycle; import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.Message; +import org.springframework.util.StringUtils; /** * @author Soby Chacko @@ -187,7 +188,10 @@ public class KafkaProducerContext implements SmartLifecycle, NamedComponen start(); } - ProducerConfiguration producerConfiguration = getTopicConfiguration(topic); + // only try to look up for a producer configuration if the topic is passed as argument + // if no topic is configured, then we'll fall back to the default if a single + // producer configuration is available + ProducerConfiguration producerConfiguration = StringUtils.hasText(topic) ? getTopicConfiguration(topic) : null; if (producerConfiguration != null) { producerConfiguration.send(topic, messageKey, message); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java index b7bdbc38a8..06b3567597 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java @@ -20,6 +20,7 @@ import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.endsWith; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.hasItem; +import static org.hamcrest.Matchers.hasSize; import static org.junit.Assert.assertThat; import java.util.ArrayList; @@ -112,13 +113,13 @@ public class OutboundTests { final List payloads = new ArrayList(); final CountDownLatch latch = new CountDownLatch(expectedMessageCount); kafkaMessageListenerContainer.setMessageListener(new MessageListener() { - + @Override public void onMessage(KafkaMessage message) { payloads.add(MessageUtils.decodePayload(message, decoder)); latch.countDown(); } - + }); kafkaMessageListenerContainer.start(); @@ -182,14 +183,14 @@ public class OutboundTests { final MutableMultimap payloadsByTopic = Multimaps.mutable.list.with(); final CountDownLatch latch = new CountDownLatch(expectedMessageCount); kafkaMessageListenerContainer.setMessageListener(new MessageListener() { - + @Override public void onMessage(KafkaMessage message) { payloadsByTopic.put(message.getMetadata().getPartition().getTopic(), MessageUtils.decodePayload(message, decoder)); latch.countDown(); } - + }); kafkaMessageListenerContainer.start(); @@ -237,12 +238,67 @@ public class OutboundTests { kafkaMessageListenerContainer.stop(); } + + @Test + public void testNoHeader() throws Exception { + + // create the topic + + try { + TopicUtils.ensureTopicCreated(kafkaRule.getZookeeperConnectionString(), TOPIC, 1, 1); + } + catch (TopicExistsException e) { + // do nothing + } + + final String suffix = UUID.randomUUID().toString(); + + KafkaMessageListenerContainer kafkaMessageListenerContainer = createMessageListenerContainer(TOPIC); + + final Decoder decoder = new StringDecoder(); + + int expectedMessageCount = 1; + final MutableMultimap payloadsByTopic = Multimaps.mutable.list.with(); + final CountDownLatch latch = new CountDownLatch(expectedMessageCount); + kafkaMessageListenerContainer.setMessageListener(new MessageListener() { + + @Override + public void onMessage(KafkaMessage message) { + payloadsByTopic.put(message.getMetadata().getPartition().getTopic(), + MessageUtils.decodePayload(message, decoder)); + latch.countDown(); + } + + }); + + kafkaMessageListenerContainer.start(); + + KafkaProducerContext producerContext = createProducerContext(); + KafkaProducerMessageHandler handler + = new KafkaProducerMessageHandler(producerContext); + + handler.handleMessage(MessageBuilder.withPayload("fooTopic1" + suffix).build()); + + StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); + evaluationContext.addPropertyAccessor(new MapAccessor()); + handler.setIntegrationEvaluationContext(evaluationContext); + + producerContext.stop(); + + latch.await(1000, TimeUnit.MILLISECONDS); + assertThat(latch.getCount(), equalTo(0L)); + assertThat(payloadsByTopic.keysView(), hasItem(TOPIC)); + assertThat(payloadsByTopic.toMap().get(TOPIC).toList(), hasSize(1)); + assertThat(payloadsByTopic.toMap().get(TOPIC), contains("fooTopic1" + suffix)); + kafkaMessageListenerContainer.stop(); + } + private KafkaMessageListenerContainer createMessageListenerContainer(String... topics) throws Exception { - ZookeeperConfiguration configuration = + ZookeeperConfiguration configuration = new ZookeeperConfiguration(new ZookeeperConnect(kafkaRule.getZookeeperConnectionString())); DefaultConnectionFactory connectionFactory = new DefaultConnectionFactory(configuration); connectionFactory.afterPropertiesSet(); - final KafkaMessageListenerContainer kafkaMessageListenerContainer = + final KafkaMessageListenerContainer kafkaMessageListenerContainer = new KafkaMessageListenerContainer(connectionFactory, topics); kafkaMessageListenerContainer.setMaxFetch(100); kafkaMessageListenerContainer.setConcurrency(1); @@ -271,5 +327,5 @@ public class OutboundTests { kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(TOPIC, config)); return kafkaProducerContext; } - + }