diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java index c166e012da..b3638eac92 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 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. @@ -32,6 +32,7 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle * @author Soby Chacko * @author Artem Bilan * @author Gary Russell + * @author Biju Kunjummen * @since 0.5 * */ @@ -75,6 +76,13 @@ public class KafkaOutboundChannelAdapterParser extends AbstractOutboundChannelAd kafkaProducerMessageHandlerBuilder.addPropertyValue("sendTimeoutExpression", sendTimeoutExpressionDef); } + BeanDefinition timestampExpressionDef = + IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("timestamp-expression", element); + + if (timestampExpressionDef != null) { + kafkaProducerMessageHandlerBuilder.addPropertyValue("timestampExpression", timestampExpressionDef); + } + return kafkaProducerMessageHandlerBuilder.getBeanDefinition(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index 74824d9754..4df1a1702b 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -40,6 +40,7 @@ import org.springframework.messaging.Message; * @param the value type. * * @author Artem Bilan + * @author Biju Kunjummen * * @since 3.0 */ @@ -194,6 +195,46 @@ public class KafkaProducerMessageHandlerSpec return _this(); } + /** + * Configure a SpEL expression to determine the timestamp at runtime against a + * request Message as a root object of evaluation context. + * @param timestampExpression the timestamp expression to use. + * @return the spec. + */ + public KafkaProducerMessageHandlerSpec timestampExpression(String timestampExpression) { + return this.timestampExpression(PARSER.parseExpression(timestampExpression)); + } + + /** + * Configure a {@link Function} that will be invoked at run time to determine the Kafka record timestamp + * will be stored in the topic. Typically used with a Java 8 Lambda expression: + * 
+	 * {@code
+	 * .timestamp(m -> m.getHeaders().get("mytimestamp_header", Long.class))
+	 * }
+	 *
+ * @param timestampFunction the partitionId function. + * @param

the expected payload type. + * @return the spec. + */ + public

KafkaProducerMessageHandlerSpec timestamp(Function, Long> timestampFunction) { + return timestampExpression(new FunctionExpression<>(timestampFunction)); + } + + /** + * Configure an {@link Expression} to determine the timestamp at runtime against a + * request Message as a root object of evaluation context. + * @param timestampExpression the timestamp expression to use. + * @return the spec. + * + * @since 3.0 + */ + public KafkaProducerMessageHandlerSpec timestampExpression(Expression timestampExpression) { + this.target.setTimestampExpression(timestampExpression); + return _this(); + } + + /** * A {@code boolean} indicating if the {@link KafkaProducerMessageHandler} * should wait for the send operation results or not. Defaults to {@code false}. diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 58e0f15fff..db44cca239 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 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. @@ -43,6 +43,8 @@ import org.springframework.util.concurrent.ListenableFuture; * @author Artem Bilan * @author Gary Russell * @author Marius Bogoevici + * @author Biju Kunjummen + * * @since 0.5 */ public class KafkaProducerMessageHandler extends AbstractMessageHandler { @@ -59,6 +61,8 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { private volatile Expression partitionIdExpression; + private volatile Expression timestampExpression; + private boolean sync; private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT); @@ -80,6 +84,18 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { this.partitionIdExpression = partitionIdExpression; } + /** + * Specify a SpEL expression to evaluate a timestamp that will be added in the Kafka record. + * The resulting value should be a {@link Long} type representing epoch time in milliseconds. + * + * @param timestampExpression the {@link Expression} for timestamp to wait for result + * fo send operation. + * @since 3.0.0 + */ + public void setTimestampExpression(Expression timestampExpression) { + this.timestampExpression = timestampExpression; + } + public KafkaTemplate getKafkaTemplate() { return this.kafkaTemplate; } @@ -142,28 +158,17 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler { ? this.messageKeyExpression.getValue(this.evaluationContext, message) : message.getHeaders().get(KafkaHeaders.MESSAGE_KEY); - ListenableFuture future; + Long timestamp = this.timestampExpression != null + ? this.timestampExpression.getValue(this.evaluationContext, message, Long.class) + : message.getHeaders().get(KafkaHeaders.TIMESTAMP, Long.class); V payload = (V) message.getPayload(); if (payload instanceof KafkaNull) { payload = null; } - if (partitionId == null) { - if (messageKey == null) { - future = this.kafkaTemplate.send(topic, payload); - } - else { - future = this.kafkaTemplate.send(topic, (K) messageKey, payload); - } - } - else { - if (messageKey == null) { - future = this.kafkaTemplate.send(topic, partitionId, payload); - } - else { - future = this.kafkaTemplate.send(topic, partitionId, (K) messageKey, payload); - } - } + + ListenableFuture future = this.kafkaTemplate.send(topic, partitionId, timestamp, (K) messageKey, payload); + if (this.sync) { Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class); if (sendTimeout == null || sendTimeout < 0) { diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd index a04f42f661..8116abc983 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd @@ -96,6 +96,14 @@ ]]> + + + + + + send-timeout-expression="1000" + timestamp-expression="T(System).currentTimeMillis()" + > diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index dbf468cc8d..4c1b8b525f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 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. @@ -48,6 +48,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Soby Chacko * @author Artem Bilan * @author Gary Russell + * @author Biju Kunjummen + * * @since 0.5 */ @RunWith(SpringJUnit4ClassRunner.class) @@ -69,6 +71,8 @@ public class KafkaOutboundAdapterParserTests { assertThat(TestUtils.getPropertyValue(messageHandler, "partitionIdExpression.expression")).isEqualTo("'2'"); assertThat(TestUtils.getPropertyValue(messageHandler, "sync", Boolean.class)).isTrue(); assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.expression")).isEqualTo("1000"); + assertThat(TestUtils.getPropertyValue(messageHandler, "timestampExpression.expression")) + .isEqualTo("T(System).currentTimeMillis()"); messageHandler = this.appContext.getBean("kafkaOutboundChannelAdapter2.handler", KafkaProducerMessageHandler.class); @@ -77,6 +81,8 @@ public class KafkaOutboundAdapterParserTests { assertThat(TestUtils.getPropertyValue(messageHandler, "sync", Boolean.class)).isFalse(); assertThat(TestUtils.getPropertyValue(messageHandler, "sendTimeoutExpression.literalValue")).isEqualTo("500"); + + } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java similarity index 66% rename from spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java rename to spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 624b6085ec..9f3880ea11 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -63,36 +63,40 @@ import org.springframework.test.context.junit4.SpringRunner; /** * @author Artem Bilan * @author Nasko Vasilev + * @author Biju Kunjummen * * @since 3.0 */ @RunWith(SpringRunner.class) @DirtiesContext -public class KafkaTests { +public class KafkaDslTests { - private static final String TEST_TOPIC = "test-topic"; + private static final String TEST_TOPIC1 = "test-topic1"; private static final String TEST_TOPIC2 = "test-topic2"; private static final String TEST_TOPIC3 = "test-topic3"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC, TEST_TOPIC2, TEST_TOPIC3); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3); @Autowired @Qualifier("sendToKafkaFlow.input") private MessageChannel sendToKafkaFlowInput; @Autowired - private PollableChannel listeningFromKafkaResults; + private PollableChannel listeningFromKafkaResults1; @Autowired - @Qualifier("kafkaProducer.handler") - private KafkaProducerMessageHandler kafkaProducer; + private PollableChannel listeningFromKafkaResults2; @Autowired - @Qualifier("kafkaProducer3.handler") - private KafkaProducerMessageHandler kafkaProducer3; + @Qualifier("kafkaProducer1.handler") + private KafkaProducerMessageHandler kafkaProducer1; + + @Autowired + @Qualifier("kafkaProducer2.handler") + private KafkaProducerMessageHandler kafkaProducer2; @Autowired private PollableChannel errorChannel; @@ -103,29 +107,47 @@ public class KafkaTests { assertThatThrownBy(() -> this.sendToKafkaFlowInput.send(new GenericMessage<>("foo"))) .hasMessageContaining("10 is not in the range"); - this.kafkaProducer.setPartitionIdExpression(new ValueExpression<>(0)); - this.kafkaProducer3.setPartitionIdExpression(new ValueExpression<>(0)); + this.kafkaProducer1.setPartitionIdExpression(new ValueExpression<>(0)); + this.kafkaProducer2.setPartitionIdExpression(new ValueExpression<>(0)); this.sendToKafkaFlowInput.send(new GenericMessage<>("foo")); for (int i = 0; i < 100; i++) { - Message receive = this.listeningFromKafkaResults.receive(20000); + Message receive = this.listeningFromKafkaResults1.receive(20000); assertThat(receive).isNotNull(); assertThat(receive.getPayload()).isEqualTo("FOO"); MessageHeaders headers = receive.getHeaders(); assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue(); Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); acknowledgment.acknowledge(); - assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC1); assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(i + 1); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo((long) i); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048633L); + } + + for (int i = 0; i < 100; i++) { + Message receive = this.listeningFromKafkaResults2.receive(20000); + assertThat(receive).isNotNull(); + assertThat(receive.getPayload()).isEqualTo("FOO"); + MessageHeaders headers = receive.getHeaders(); + assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue(); + Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); + acknowledgment.acknowledge(); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(TEST_TOPIC2); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(i + 1); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo((long) i); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048644L); } Message message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build(); this.sendToKafkaFlowInput.send(message); - assertThat(this.listeningFromKafkaResults.receive(10)).isNull(); + assertThat(this.listeningFromKafkaResults1.receive(10)).isNull(); Message error = this.errorChannel.receive(10000); assertThat(error).isNotNull(); @@ -151,10 +173,10 @@ public class KafkaTests { } @Bean - public IntegrationFlow listeningFromKafkaFlow() { + public IntegrationFlow topic1ListenerFromKafkaFlow() { return IntegrationFlows .from(Kafka.messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC) + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) .configureListenerContainer(c -> c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)) .errorChannel("errorChannel") @@ -164,7 +186,25 @@ public class KafkaTests { m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101, f -> f.throwExceptionOnRejection(true)) .transform(String::toUpperCase) - .channel(c -> c.queue("listeningFromKafkaResults")) + .channel(c -> c.queue("listeningFromKafkaResults1")) + .get(); + } + + @Bean + public IntegrationFlow topic2ListenerFromKafkaFlow() { + return IntegrationFlows + .from(Kafka.messageDrivenChannelAdapter(consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2) + .configureListenerContainer(c -> + c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)) + .errorChannel("errorChannel") + .retryTemplate(new RetryTemplate()) + .filterInRetry(true)) + .filter(Message.class, m -> + m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101, + f -> f.throwExceptionOnRejection(true)) + .transform(String::toUpperCase) + .channel(c -> c.queue("listeningFromKafkaResults2")) .get(); } @@ -178,15 +218,18 @@ public class KafkaTests { return f -> f .split(p -> Stream.generate(() -> p).limit(101).iterator(), null) .publishSubscribeChannel(c -> c - .subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC), - e -> e.id("kafkaProducer"))) - .subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC3), - e -> e.id("kafkaProducer3"))) + .subscribe(sf -> sf.handle( + kafkaMessageHandler(producerFactory(), TEST_TOPIC1) + .timestampExpression("T(Long).valueOf('1487694048633')"), + e -> e.id("kafkaProducer1"))) + .subscribe(sf -> sf.handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2) + .timestamp(m -> 1487694048644L), + e -> e.id("kafkaProducer2"))) ); } - private KafkaProducerMessageHandlerSpec - kafkaMessageHandler(ProducerFactory producerFactory, String topic) { + private KafkaProducerMessageHandlerSpec kafkaMessageHandler( + ProducerFactory producerFactory, String topic) { return Kafka .outboundChannelAdapter(producerFactory) .messageKey(m -> m diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 5d1cb45773..c24fd42424 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -19,6 +19,7 @@ package org.springframework.integration.kafka.inbound; import static org.assertj.core.api.Assertions.assertThat; import java.lang.reflect.Type; +import java.util.Arrays; import java.util.List; import java.util.Map; @@ -54,8 +55,11 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.PollableChannel; /** + * * @author Gary Russell * @author Artem Bilan + * @author Biju Kunjummen + * * @since 2.0 * */ @@ -95,10 +99,10 @@ public class MessageDrivenAdapterTests { ContainerTestUtils.waitForAssignment(container, 2); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic1); - template.sendDefault(1, "foo"); + template.sendDefault(0, 1487694048607L, 1, "foo"); Message received = out.receive(10000); assertThat(received).isNotNull(); @@ -108,6 +112,9 @@ public class MessageDrivenAdapterTests { assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic1); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get("testHeader")).isEqualTo("testValue"); template.sendDefault(1, null); @@ -121,6 +128,9 @@ public class MessageDrivenAdapterTests { assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic1); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(1L); + assertThat((Long) headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isGreaterThan(0L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(headers.get("testHeader")).isEqualTo("testValue"); adapter.setMessageConverter(new RecordMessageConverter() { @@ -176,8 +186,8 @@ public class MessageDrivenAdapterTests { ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic2); - template.sendDefault(1, "foo"); - template.sendDefault(1, "bar"); + template.sendDefault(0, 1487694048607L, 1, "foo"); + template.sendDefault(0, 1487694048608L, 1, "bar"); Message received = out.receive(10000); assertThat(received).isNotNull(); @@ -187,10 +197,14 @@ public class MessageDrivenAdapterTests { assertThat(list.size()).isGreaterThan(0); MessageHeaders headers = received.getHeaders(); - assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY).toString()).contains("[1"); - assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC).toString()).contains(topic2); - assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID).toString()).contains("0"); - assertThat(headers.get(KafkaHeaders.OFFSET).toString()).contains("[0"); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(Arrays.asList(1, 1)); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(Arrays.asList("testTopic2", "testTopic2")); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(Arrays.asList(0, 0)); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(Arrays.asList(0L, 1L)); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)) + .isEqualTo(Arrays.asList("CREATE_TIME", "CREATE_TIME")); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)) + .isEqualTo(Arrays.asList(1487694048607L, 1487694048608L)); assertThat(headers.get("testHeader")).isEqualTo("testValue"); adapter.setMessageConverter(new BatchMessageConverter() { @@ -239,7 +253,7 @@ public class MessageDrivenAdapterTests { ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic3); - template.sendDefault(1, "{\"bar\":\"baz\"}"); + template.sendDefault(0, 1487694048607L, 1, "{\"bar\":\"baz\"}"); Message received = out.receive(10000); assertThat(received).isNotNull(); @@ -249,6 +263,9 @@ public class MessageDrivenAdapterTests { assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic3); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L); + + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); assertThat(received.getPayload()).isInstanceOf(Map.class); adapter.setPayloadType(Foo.class); @@ -262,6 +279,9 @@ public class MessageDrivenAdapterTests { assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic3); assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0); assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(1L); + assertThat((Long) headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isGreaterThan(0L); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(received.getPayload()).isInstanceOf(Foo.class); assertThat(received.getPayload()).isEqualTo(new Foo("baz")); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index dcc4c3cfb8..dd81f55049 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; import static org.springframework.kafka.test.assertj.KafkaConditions.key; import static org.springframework.kafka.test.assertj.KafkaConditions.partition; +import static org.springframework.kafka.test.assertj.KafkaConditions.timestamp; import static org.springframework.kafka.test.assertj.KafkaConditions.value; import org.apache.kafka.clients.consumer.Consumer; @@ -30,6 +31,8 @@ import org.junit.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; @@ -44,8 +47,9 @@ import org.springframework.messaging.Message; /** * @author Gary Russell - * @since 2.0 + * @author Biju Kunjummen * + * @since 2.0 */ public class KafkaProducerMessageHandlerTests { @@ -120,4 +124,65 @@ public class KafkaProducerMessageHandlerTests { assertThat(record.value()).isNull(); } + @Test + public void testOutboundWithTimestamp() { + ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + KafkaTestUtils.producerProps(embeddedKafka)); + KafkaTemplate template = new KafkaTemplate<>(producerFactory); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic1) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .setHeader(KafkaHeaders.TIMESTAMP, 1487694048607L) + .build(); + handler.handleMessage(message); + + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic1); + assertThat(record).has(key(2)); + assertThat(record).has(partition(1)); + assertThat(record).has(value("foo")); + assertThat(record).has(timestamp(1487694048607L)); + } + + @Test + public void testOutboundWithTimestampExpression() { + ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + KafkaTestUtils.producerProps(embeddedKafka)); + KafkaTemplate template = new KafkaTemplate<>(producerFactory); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); + + Message message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic1) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .build(); + + handler.setTimestampExpression(new ValueExpression<>(1487694048633L)); + + handler.handleMessage(message); + + ConsumerRecord record1 = KafkaTestUtils.getSingleRecord(consumer, topic1); + assertThat(record1).has(key(2)); + assertThat(record1).has(partition(1)); + assertThat(record1).has(value("foo")); + assertThat(record1).has(timestamp(1487694048633L)); + + Long currentTimeMarker = System.currentTimeMillis(); + handler.setTimestampExpression(new FunctionExpression>(m -> System.currentTimeMillis())); + + handler.handleMessage(message); + + ConsumerRecord record2 = KafkaTestUtils.getSingleRecord(consumer, topic1); + assertThat(record2).has(key(2)); + assertThat(record2).has(partition(1)); + assertThat(record2).has(value("foo")); + assertThat(record2.timestamp()).isGreaterThanOrEqualTo(currentTimeMarker); + } + }