Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-158: Add record timestamp support

Browse Source 
Fixes spring-projects/spring-integration-kafka#158

Fixing build - replaced a call to removed API in spring-kafka project

Support for timestamp expression in SI schema and SI DSL

* Polishing according PR comments
This commit is contained in:
Biju Kunjummen
2017-02-22 14:45:37 -08:00
committed by Artem Bilan
parent 328e9cdb57
commit 9e047e7703
9 changed files with 255 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaOutboundChannelAdapterParser.java
View File

@@ -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();
}

43
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java
View File

@@ -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 <V> the value type.
*
* @author Artem Bilan
* @author Biju Kunjummen
*
* @since 3.0
*/
@@ -194,6 +195,46 @@ public class KafkaProducerMessageHandlerSpec<K, V>
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<K, V> 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:
* <pre class="code">
* {@code
* .timestamp(m -> m.getHeaders().get("mytimestamp_header", Long.class))
* }
* </pre>
* @param timestampFunction the partitionId function.
* @param <P> the expected payload type.
* @return the spec.
*/
public <P> KafkaProducerMessageHandlerSpec<K, V> timestamp(Function<Message<P>, 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<K, V> 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}.

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

@@ -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<K, V> extends AbstractMessageHandler {
@@ -59,6 +61,8 @@ public class KafkaProducerMessageHandler<K, V> 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<K, V> 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<K, V> 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) {

8
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-3.0.xsd
View File

@@ -96,6 +96,14 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="timestamp-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the timestamp for a Kafka record
against the Message at runtime.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="sync">
<xsd:annotation>
<xsd:documentation><![CDATA[

4
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml
View File

@@ -18,7 +18,9 @@
topic="foo"
message-key-expression="'bar'"
partition-id-expression="'2'"
send-timeout-expression="1000">
send-timeout-expression="1000"
timestamp-expression="T(System).currentTimeMillis()"
>
<int-kafka:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice" />
</int-kafka:request-handler-advice-chain>

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java
View File

@@ -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");
}

89
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaTests.java → spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -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<String> 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))
.<String, String>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))
.<String, String>transform(String::toUpperCase)
.channel(c -> c.queue("listeningFromKafkaResults2"))
.get();
}
@@ -178,15 +218,18 @@ public class KafkaTests {
return f -> f
.<String>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<Integer, String>
kafkaMessageHandler(ProducerFactory<Integer, String> producerFactory, String topic) {
private KafkaProducerMessageHandlerSpec<Integer, String> kafkaMessageHandler(
ProducerFactory<Integer, String> producerFactory, String topic) {
return Kafka
.outboundChannelAdapter(producerFactory)
.messageKey(m -> m

40
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> 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<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> 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<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> 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"));

69
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -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<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> 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<Integer, String> 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<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> 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<Integer, String> 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<Message<?>>(m -> System.currentTimeMillis()));
handler.handleMessage(message);
ConsumerRecord<Integer, String> 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);
}
}