splitter: 1.0.0-spring-ingetration - using a `message-key-expression`

[demos/kafka/wordcount] / src / main / java / de / juplo / kafka / wordcount / splitter / SplitterApplication.java

diff --git a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
index 412f429..246fffd 100644 (file)
--- a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
+++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplication.java
@@ -4,7 +4,10 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.common.LiteralExpression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.annotation.InboundChannelAdapter;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.config.EnableIntegration;
@@ -39,6 +42,9 @@ public class SplitterApplication
                KafkaProducerMessageHandler<String, String> handler =
                                new KafkaProducerMessageHandler<>(kafkaTemplate);
                handler.setTopicExpression(new LiteralExpression(properties.getOutputTopic()));
+               final ExpressionParser parser = new SpelExpressionParser();
+               Expression expression = parser.parseExpression("headers['kafka_receivedMessageKey']");
+               handler.setMessageKeyExpression(expression);
                return handler;
        }