From 30b387cc78b0766db1500698cd439d086252b71b Mon Sep 17 00:00:00 2001 From: Biju Kunjummen Date: Tue, 21 Feb 2017 09:24:52 -0800 Subject: [PATCH] Support for timestamps in Kafka Template Support for mapping Kafka timestamp and timestamp type to message headers and back Incorporated review comments, updated documentation Updated documentation for storing/retrieving timestamps * Polishing according PR comments Conflicts: src/reference/asciidoc/whats-new.adoc --- gradle/wrapper/gradle-wrapper.jar | Bin 54208 -> 54208 bytes gradle/wrapper/gradle-wrapper.properties | 2 +- .../kafka/test/assertj/KafkaConditions.java | 43 +++++++- .../kafka/test/hamcrest/KafkaMatchers.java | 73 ++++++++++++-- .../test/hamcrest/KafkaMatchersTests.java | 63 ++++++++++++ .../kafka/core/KafkaOperations.java | 47 ++++++--- .../kafka/core/KafkaTemplate.java | 16 ++- .../kafka/support/KafkaHeaders.java | 18 +++- .../BatchMessagingMessageConverter.java | 9 +- .../converter/MessagingMessageConverter.java | 8 +- .../kafka/core/KafkaTemplateTests.java | 93 +++++++++++++++++- .../converter/BatchMessageConverterTests.java | 71 +++++++++++++ src/checkstyle/checkstyle-suppressions.xml | 1 + src/reference/asciidoc/appendix.adoc | 3 + src/reference/asciidoc/changes-1.0-1.1.adoc | 25 +++++ src/reference/asciidoc/index-docinfo.xml | 4 + src/reference/asciidoc/kafka.adoc | 36 +++++-- src/reference/asciidoc/testing.adoc | 38 +++++++ src/reference/asciidoc/whats-new.adoc | 32 ++---- 19 files changed, 514 insertions(+), 68 deletions(-) create mode 100644 spring-kafka-test/src/test/java/org/springframework/kafka/test/hamcrest/KafkaMatchersTests.java create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java create mode 100644 src/reference/asciidoc/changes-1.0-1.1.adoc diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar index c89470b387752698b2422dea30156c056e59a3f9..cb19da3c5554530f50d1b9659f430d5a7623e912 100644 GIT binary patch delta 26 gcmX@GocX|V<_*z@m}^qPHpd+j5(F_O7hm=T0G#d%FaQ7m delta 26 gcmX@GocX|V<_*z@m^TNjY>qo5BnV(value); } + /** + * @param value the timestamp. + * @return a Condition that matches the timestamp value in a consumer record. + * @since 2.0 + */ + public static Condition> timestamp(long value) { + return timestamp(TimestampType.CREATE_TIME, value); + } + + /** + * @param type the type of timestamp + * @param value the timestamp. + * @return a Condition that matches the timestamp value in a consumer record. + * @since 2.0 + */ + public static Condition> timestamp(TimestampType type, long value) { + return new ConsumerRecordTimestampCondition(type, value); + } + /** * @param partition the partition. * @return a Condition that matches the partition in a consumer record. @@ -90,6 +111,26 @@ public final class KafkaConditions { } + public static class ConsumerRecordTimestampCondition extends Condition> { + + private final TimestampType type; + + private final long ts; + + public ConsumerRecordTimestampCondition(TimestampType type, long ts) { + super("a ConsumerRecord with timestamp of type: " + type + " and timestamp: " + ts); + this.type = type; + this.ts = ts; + } + + @Override + public boolean matches(ConsumerRecord value) { + return value != null && + (value.timestampType() == this.type && value.timestamp() == this.ts); + } + + } + public static class ConsumerRecordPartitionCondition extends Condition> { private final int partition; diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/hamcrest/KafkaMatchers.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/hamcrest/KafkaMatchers.java index 3aa03824..b0f69913 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/hamcrest/KafkaMatchers.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/hamcrest/KafkaMatchers.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. @@ -17,6 +17,7 @@ package org.springframework.kafka.test.hamcrest; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.record.TimestampType; import org.hamcrest.Description; import org.hamcrest.DiagnosingMatcher; import org.hamcrest.Matcher; @@ -25,6 +26,7 @@ import org.hamcrest.Matcher; * Hamcrest {@link Matcher}s utilities. * * @author Gary Russell + * @author Biju Kunjummen * */ public final class KafkaMatchers { @@ -59,8 +61,30 @@ public final class KafkaMatchers { return new ConsumerRecordPartitionMatcher(partition); } + /** + * Matcher testing the timestamp of a {@link ConsumerRecord} asssuming the topic has been set with + * {@link org.apache.kafka.common.record.TimestampType#CREATE_TIME CreateTime}. + * @param ts timestamp of the consumer record. + * @return a Matcher that matches the timestamp in a consumer record. + * @since 2.0 + */ + public static Matcher> hasTimestamp(long ts) { + return hasTimestamp(TimestampType.CREATE_TIME, ts); + } - public static class ConsumerRecordKeyMatcher extends DiagnosingMatcher> { + /** + * Matcher testing the timestamp of a {@link ConsumerRecord} + * @param type timestamp type of the record + * @param ts timestamp of the consumer record. + * @return a Matcher that matches the timestamp in a consumer record. + * @since 2.0 + */ + public static Matcher> hasTimestamp(TimestampType type, long ts) { + return new ConsumerRecordTimestampMatcher(type, ts); + } + + public static class ConsumerRecordKeyMatcher + extends DiagnosingMatcher> { private final K key; @@ -70,7 +94,8 @@ public final class KafkaMatchers { @Override public void describeTo(Description description) { - description.appendText("a ConsumerRecord with key ").appendText(this.key.toString()); + description.appendText("a ConsumerRecord with key ") + .appendText(this.key.toString()); } @Override @@ -78,7 +103,8 @@ public final class KafkaMatchers { @SuppressWarnings("unchecked") ConsumerRecord record = (ConsumerRecord) item; boolean matches = record != null - && ((record.key() == null && this.key == null) || record.key().equals(this.key)); + && ((record.key() == null && this.key == null) + || record.key().equals(this.key)); if (!matches) { mismatchDescription.appendText("is ").appendValue(record); } @@ -97,7 +123,8 @@ public final class KafkaMatchers { @Override public void describeTo(Description description) { - description.appendText("a ConsumerRecord with value ").appendText(this.payload.toString()); + description.appendText("a ConsumerRecord with value ") + .appendText(this.payload.toString()); } @Override @@ -123,7 +150,8 @@ public final class KafkaMatchers { @Override public void describeTo(Description description) { - description.appendText("a ConsumerRecord with partition ").appendValue(this.partition); + description.appendText("a ConsumerRecord with partition ") + .appendValue(this.partition); } @Override @@ -139,4 +167,37 @@ public final class KafkaMatchers { } + public static class ConsumerRecordTimestampMatcher extends DiagnosingMatcher> { + + private final TimestampType type; + + private final long ts; + + public ConsumerRecordTimestampMatcher(TimestampType type, long ts) { + this.type = type; + this.ts = ts; + } + + @Override + protected boolean matches(Object item, Description mismatchDescription) { + @SuppressWarnings("unchecked") + ConsumerRecord record = (ConsumerRecord) item; + + boolean matches = record != null && + (record.timestampType() == this.type && record.timestamp() == this.ts); + + if (!matches) { + mismatchDescription.appendText("is ").appendValue(record); + } + return matches; + } + + @Override + public void describeTo(Description description) { + description.appendText("a ConsumerRecord with timestamp of type: ") + .appendValue(this.type).appendText(" and value: ").appendValue(this.ts); + } + + } + } diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/hamcrest/KafkaMatchersTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/hamcrest/KafkaMatchersTests.java new file mode 100644 index 00000000..a4484ab8 --- /dev/null +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/hamcrest/KafkaMatchersTests.java @@ -0,0 +1,63 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.test.hamcrest; + +import static org.junit.Assert.assertThat; +import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasKey; +import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasPartition; +import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasTimestamp; +import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasValue; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.record.TimestampType; +import org.hamcrest.Matchers; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +/** + * @author Biju Kunjummen + * + * @since 2.0 + */ +public class KafkaMatchersTests { + + @Rule + public ExpectedException expectedException = ExpectedException.none(); + + @Test + public void testKeyMatcher() { + ConsumerRecord record = new ConsumerRecord<>("topic", 0, 10, + 1487694048607L, TimestampType.CREATE_TIME, 123L, 2, 3, "key1", "value1"); + assertThat(record, hasKey("key1")); + assertThat(record, hasValue("value1")); + assertThat(record, hasPartition(0)); + assertThat(record, hasTimestamp(1487694048607L)); + assertThat(record, hasTimestamp(TimestampType.CREATE_TIME, 1487694048607L)); + } + + @Test + public void noMatchOnTimestamp() { + ConsumerRecord record = new ConsumerRecord<>("topic", 0, 10, + 1487694048607L, TimestampType.CREATE_TIME, 123L, 2, 3, "key1", "value1"); + + expectedException.expectMessage(Matchers.containsString( + "Expected: a ConsumerRecord with timestamp of type: and value: <123L>")); + assertThat(record, hasTimestamp(123L)); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java index 8ba639ae..f4a69021 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.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. @@ -34,8 +34,17 @@ import org.springframework.util.concurrent.ListenableFuture; * @param the key type. * @param the value type. * + * If the Kafka topic is set with {@link org.apache.kafka.common.record.TimestampType#CREATE_TIME CreateTime} + * all send operations will use the user provided time if provided, else + * {@link org.apache.kafka.clients.producer.KafkaProducer} will generate one + * + * If the topic is set with {@link org.apache.kafka.common.record.TimestampType#LOG_APPEND_TIME LogAppendTime} + * then the user provided timestamp will be ignored and instead will be the + * Kafka broker local time when the message is appended + * * @author Marius Bogoevici * @author Gary Russell + * @author Biju Kunjummen */ public interface KafkaOperations { @@ -61,7 +70,18 @@ public interface KafkaOperations { * @param data the data. * @return a Future for the {@link SendResult}. */ - ListenableFuture> sendDefault(int partition, K key, V data); + ListenableFuture> sendDefault(Integer partition, K key, V data); + + /** + * Send the data to the default topic with the provided key and partition. + * @param partition the partition. + * @param timestamp the timestamp of the record. + * @param key the key. + * @param data the data. + * @return a Future for the {@link SendResult}. + * @since 2.0 + */ + ListenableFuture> sendDefault(Integer partition, Long timestamp, K key, V data); /** * Send the data to the provided topic with no key or partition. @@ -80,15 +100,6 @@ public interface KafkaOperations { */ ListenableFuture> send(String topic, K key, V data); - /** - * Send the data to the provided topic with the provided partition and no key. - * @param topic the topic. - * @param partition the partition. - * @param data The data. - * @return a Future for the {@link SendResult}. - */ - ListenableFuture> send(String topic, int partition, V data); - /** * Send the data to the provided topic with the provided key and partition. * @param topic the topic. @@ -97,7 +108,19 @@ public interface KafkaOperations { * @param data the data. * @return a Future for the {@link SendResult}. */ - ListenableFuture> send(String topic, int partition, K key, V data); + ListenableFuture> send(String topic, Integer partition, K key, V data); + + /** + * Send the data to the provided topic with the provided key and partition. + * @param topic the topic. + * @param partition the partition. + * @param timestamp the timestamp of the record. + * @param key the key. + * @param data the data. + * @return a Future for the {@link SendResult}. + * @since 2.0 + */ + ListenableFuture> send(String topic, Integer partition, Long timestamp, K key, V data); /** * Send a message with routing information in message headers. The message payload diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 197c57d4..81ed165a 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -50,6 +50,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; * @author Gary Russell * @author Igor Stepanov * @author Artem Bilan + * @author Biju Kunjummen */ public class KafkaTemplate implements KafkaOperations { @@ -141,10 +142,15 @@ public class KafkaTemplate implements KafkaOperations { } @Override - public ListenableFuture> sendDefault(int partition, K key, V data) { + public ListenableFuture> sendDefault(Integer partition, K key, V data) { return send(this.defaultTopic, partition, key, data); } + @Override + public ListenableFuture> sendDefault(Integer partition, Long timestamp, K key, V data) { + return send(this.defaultTopic, partition, timestamp, key, data); + } + @Override public ListenableFuture> send(String topic, V data) { ProducerRecord producerRecord = new ProducerRecord<>(topic, data); @@ -158,14 +164,14 @@ public class KafkaTemplate implements KafkaOperations { } @Override - public ListenableFuture> send(String topic, int partition, V data) { - ProducerRecord producerRecord = new ProducerRecord(topic, partition, null, data); + public ListenableFuture> send(String topic, Integer partition, K key, V data) { + ProducerRecord producerRecord = new ProducerRecord<>(topic, partition, key, data); return doSend(producerRecord); } @Override - public ListenableFuture> send(String topic, int partition, K key, V data) { - ProducerRecord producerRecord = new ProducerRecord<>(topic, partition, key, data); + public ListenableFuture> send(String topic, Integer partition, Long timestamp, K key, V data) { + ProducerRecord producerRecord = new ProducerRecord<>(topic, partition, timestamp, key, data); return doSend(producerRecord); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java b/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java index 9a5ce0b5..34eb74c7 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -22,6 +22,7 @@ package org.springframework.kafka.support; * @author Artem Bilan * @author Marius Bogoevici * @author Gary Russell + * @author Biju Kunjummen */ public abstract class KafkaHeaders { @@ -67,4 +68,19 @@ public abstract class KafkaHeaders { */ public static final String RECEIVED_PARTITION_ID = PREFIX + "receivedPartitionId"; + /** + * The header for holding the {@link org.apache.kafka.common.record.TimestampType type} of timestamp. + */ + public static final String TIMESTAMP_TYPE = PREFIX + "timestampType"; + + /** + * The header for holding the timestamp of the producer record. + */ + public static final String TIMESTAMP = PREFIX + "timestamp"; + + /** + * The header for holding the timestamp of the consumer record. + */ + public static final String RECEIVED_TIMESTAMP = PREFIX + "receivedTimestamp"; + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java index dfe35b5f..916d51c9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.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. @@ -42,6 +42,7 @@ import org.springframework.messaging.support.MessageBuilder; * @author Marius Bogoevici * @author Gary Russell * @author Dariusz Szablinski + * @author Biju Kunjummen * @since 1.1 */ public class BatchMessagingMessageConverter implements BatchMessageConverter { @@ -80,10 +81,14 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter { List topics = new ArrayList<>(); List partitions = new ArrayList<>(); List offsets = new ArrayList<>(); + List timestampTypes = new ArrayList<>(); + List timestamps = new ArrayList<>(); rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, keys); rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, topics); rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, partitions); rawHeaders.put(KafkaHeaders.OFFSET, offsets); + rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, timestampTypes); + rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, timestamps); if (acknowledgment != null) { rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment); @@ -95,6 +100,8 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter { topics.add(record.topic()); partitions.add(record.partition()); offsets.add(record.offset()); + timestampTypes.add(record.timestampType().name()); + timestamps.add(record.timestamp()); } return MessageBuilder.createMessage(payloads, kafkaMessageHeaders); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java index 53319de2..ef065eb4 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.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. @@ -39,6 +39,7 @@ import org.springframework.messaging.support.MessageBuilder; * @author Marius Bogoevici * @author Gary Russell * @author Dariusz Szablinski + * @author Biju Kunjummen */ public class MessagingMessageConverter implements RecordMessageConverter { @@ -74,6 +75,8 @@ public class MessagingMessageConverter implements RecordMessageConverter { rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, record.topic()); rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, record.partition()); rawHeaders.put(KafkaHeaders.OFFSET, record.offset()); + rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, record.timestampType().name()); + rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, record.timestamp()); if (acknowledgment != null) { rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment); @@ -90,7 +93,8 @@ public class MessagingMessageConverter implements RecordMessageConverter { Integer partition = headers.get(KafkaHeaders.PARTITION_ID, Integer.class); Object key = headers.get(KafkaHeaders.MESSAGE_KEY); Object payload = convertPayload(message); - return new ProducerRecord(topic == null ? defaultTopic : topic, partition, key, payload); + Long timestamp = headers.get(KafkaHeaders.TIMESTAMP, Long.class); + return new ProducerRecord(topic == null ? defaultTopic : topic, partition, timestamp, key, payload); } /** diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index 4d181efe..c1713d2b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.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.kafka.core; import static org.assertj.core.api.Assertions.assertThat; 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 java.util.List; @@ -45,8 +46,10 @@ import org.junit.Test; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.ProducerListenerAdapter; import org.springframework.kafka.support.SendResult; +import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; @@ -55,6 +58,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Gary Russell * @author Artem Bilan * @author Igor Stepanov + * @author Biju Kunjummen */ public class KafkaTemplateTests { @@ -70,8 +74,7 @@ public class KafkaTemplateTests { @BeforeClass public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory( - consumerProps); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); consumer = cf.createConsumer(); embeddedKafka.consumeFromAnEmbeddedTopic(consumer, INT_KEY_TOPIC); } @@ -84,26 +87,32 @@ public class KafkaTemplateTests { @Test public void testTemplate() throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic(INT_KEY_TOPIC); + template.sendDefault("foo"); assertThat(KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC)).has(value("foo")); + template.sendDefault(0, 2, "bar"); ConsumerRecord received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("bar")); + template.send(INT_KEY_TOPIC, 0, 2, "baz"); received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("baz")); - template.send(INT_KEY_TOPIC, 0, "qux"); + + template.send(INT_KEY_TOPIC, 0, null, "qux"); received = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); assertThat(received).has(key((Integer) null)); assertThat(received).has(partition(0)); assertThat(received).has(value("qux")); + template.send(MessageBuilder.withPayload("fiz") .setHeader(KafkaHeaders.TOPIC, INT_KEY_TOPIC) .setHeader(KafkaHeaders.PARTITION_ID, 0) @@ -113,6 +122,7 @@ public class KafkaTemplateTests { assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("fiz")); + template.send(MessageBuilder.withPayload("buz") .setHeader(KafkaHeaders.PARTITION_ID, 0) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) @@ -121,6 +131,7 @@ public class KafkaTemplateTests { assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("buz")); + Map metrics = template.execute(Producer::metrics); assertThat(metrics).isNotNull(); metrics = template.metrics(); @@ -131,6 +142,74 @@ public class KafkaTemplateTests { pf.destroy(); } + @Test + public void testTemplateWithTimestamps() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + + template.setDefaultTopic(INT_KEY_TOPIC); + + template.sendDefault(0, 1487694048607L, null, "foo-ts1"); + ConsumerRecord r1 = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); + assertThat(r1).has(value("foo-ts1")); + assertThat(r1).has(timestamp(1487694048607L)); + + template.send(INT_KEY_TOPIC, 0, 1487694048610L, null, "foo-ts2"); + ConsumerRecord r2 = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); + assertThat(r2).has(value("foo-ts2")); + assertThat(r2).has(timestamp(1487694048610L)); + + Map metrics = template.execute(Producer::metrics); + assertThat(metrics).isNotNull(); + metrics = template.metrics(); + assertThat(metrics).isNotNull(); + List partitions = template.partitionsFor(INT_KEY_TOPIC); + assertThat(partitions).isNotNull(); + assertThat(partitions.size()).isEqualTo(2); + pf.destroy(); + } + + @Test + public void testWithMessage() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + + Message message1 = MessageBuilder.withPayload("foo-message") + .setHeader(KafkaHeaders.TOPIC, INT_KEY_TOPIC) + .setHeader(KafkaHeaders.PARTITION_ID, 0) + .build(); + + template.send(message1); + + ConsumerRecord r1 = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); + assertThat(r1).has(value("foo-message")); + + Message message2 = MessageBuilder.withPayload("foo-message-2") + .setHeader(KafkaHeaders.TOPIC, INT_KEY_TOPIC) + .setHeader(KafkaHeaders.PARTITION_ID, 0) + .setHeader(KafkaHeaders.TIMESTAMP, 1487694048615L) + .build(); + + template.send(message2); + + ConsumerRecord r2 = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); + assertThat(r2).has(value("foo-message-2")); + assertThat(r2).has(timestamp(1487694048615L)); + + MessagingMessageConverter messageConverter = new MessagingMessageConverter(); + + Message recordToMessage = messageConverter.toMessage(r2, null, String.class); + + assertThat(recordToMessage.getHeaders().get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); + assertThat(recordToMessage.getHeaders().get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048615L); + assertThat(recordToMessage.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(INT_KEY_TOPIC); + assertThat(recordToMessage.getPayload()).isEqualTo("foo-message-2"); + + pf.destroy(); + } + @Test public void withListener() throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); @@ -155,6 +234,9 @@ public class KafkaTemplateTests { template.sendDefault("foo"); template.flush(); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + + //Drain the topic + KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); pf.destroy(); } @@ -204,6 +286,7 @@ public class KafkaTemplateTests { assertThat(record).has(Assertions.>allOf(key("foo"), value("bar"))); consumer.close(); pf.createProducer().close(); + pf.destroy(); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java new file mode 100644 index 00000000..197cc524 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java @@ -0,0 +1,71 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.support.converter; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.record.TimestampType; +import org.junit.Test; + +import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; + +/** + * @author Biju Kunjummen + * @since 2.0 + */ +public class BatchMessageConverterTests { + + @Test + public void testBatchConverters() throws Exception { + List> consumerRecords = new ArrayList<>(); + consumerRecords.add(new ConsumerRecord<>("topic1", 0, 1, 1487694048607L, + TimestampType.CREATE_TIME, 123L, 2, 3, "key1", "value1")); + consumerRecords.add(new ConsumerRecord<>("topic1", 0, 2, 1487694048608L, + TimestampType.CREATE_TIME, 123L, 2, 3, "key2", "value2")); + consumerRecords.add(new ConsumerRecord<>("topic1", 0, 3, 1487694048609L, + TimestampType.CREATE_TIME, 123L, 2, 3, "key3", "value3")); + + BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter(); + + Message message = batchMessageConverter.toMessage(consumerRecords, null, + String.class); + + assertThat(message.getPayload()) + .isEqualTo(Arrays.asList("value1", "value2", "value3")); + + MessageHeaders headers = message.getHeaders(); + assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)) + .isEqualTo(Arrays.asList("topic1", "topic1", "topic1")); + assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)) + .isEqualTo(Arrays.asList("key1", "key2", "key3")); + assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)) + .isEqualTo(Arrays.asList(0, 0, 0)); + assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(Arrays.asList(1L, 2L, 3L)); + assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)) + .isEqualTo(Arrays.asList("CREATE_TIME", "CREATE_TIME", "CREATE_TIME")); + assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)) + .isEqualTo(Arrays.asList(1487694048607L, 1487694048608L, 1487694048609L)); + } + +} diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 4097d904..05fa5cc4 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -6,4 +6,5 @@ + diff --git a/src/reference/asciidoc/appendix.adoc b/src/reference/asciidoc/appendix.adoc index f2cf08f7..efdb53a8 100644 --- a/src/reference/asciidoc/appendix.adoc +++ b/src/reference/asciidoc/appendix.adoc @@ -1 +1,4 @@ +[[history]] == Change History + +include::./changes-1.0-1.1.adoc[] \ No newline at end of file diff --git a/src/reference/asciidoc/changes-1.0-1.1.adoc b/src/reference/asciidoc/changes-1.0-1.1.adoc new file mode 100644 index 00000000..e7acc609 --- /dev/null +++ b/src/reference/asciidoc/changes-1.0-1.1.adoc @@ -0,0 +1,25 @@ +[[migration-1.0-1.1]] +=== Changes between 1.0 and 1.1 + +==== Kafka Client + +This version uses the Apache Kafka 0.10.x.x client. + +==== Batch Listeners + +Listeners can be configured to receive the entire batch of messages returned by the `consumer.poll()` operation, rather than one at a time. + +==== Null Payloads + +Null payloads are used to "delete" keys when using log compaction. + +==== Initial Offset + +When explicitly assigning partitions, you can now configure the initial offset relative to the current position for the consumer group, rather than absolute or relative to the current end. + +==== Seek + +You can now seek the position of each topic/partition. +This can be used to set the initial position during initialization when group management is in use and Kafka assigns the partitions. +You can also seek when an idle container is detected, or at any arbitrary point in your application's execution. +See <> for more information. diff --git a/src/reference/asciidoc/index-docinfo.xml b/src/reference/asciidoc/index-docinfo.xml index 1fe92791..42b5626b 100644 --- a/src/reference/asciidoc/index-docinfo.xml +++ b/src/reference/asciidoc/index-docinfo.xml @@ -13,6 +13,10 @@ Artem Bilan + + Biju + Kunjummen + diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index af11a52d..83610cf4 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -4,6 +4,7 @@ ==== Sending Messages +[[kafka-template]] ===== KafkaTemplate The `KafkaTemplate` wraps a producer and provides convenience methods to send data to kafka topics. @@ -15,15 +16,17 @@ ListenableFuture> sendDefault(V data); ListenableFuture> sendDefault(K key, V data); -ListenableFuture> sendDefault(int partition, K key, V data); +ListenableFuture> sendDefault(Integer partition, K key, V data); + +ListenableFuture> sendDefault(Integer partition, Long timestamp, K key, V data); ListenableFuture> send(String topic, V data); ListenableFuture> send(String topic, K key, V data); -ListenableFuture> send(String topic, int partition, V data); +ListenableFuture> send(String topic, Integer partition, K key, V data); -ListenableFuture> send(String topic, int partition, K key, V data); +ListenableFuture> send(String topic, Integer partition, Long timestamp, K key, V data); ListenableFuture> send(Message message); @@ -45,10 +48,15 @@ interface ProducerCallback { ---- -The first 3 methods require that a default topic has been provided to the template. +The `sendDefault` API requires that a default topic has been provided to the template. -The `metrics` and `partitionsFor` methods simply delegate to the same methods on the underlying `Producer`. -The `execute` method provides direct access to the underlying `Producer`. +The API which take in a `timestamp` as a parameter will store this timestamp in the record. +The behavior of the user provided timestamp is stored is dependent on the timestamp type configured on the Kafka topic. +If the topic is configured to use `CREATE_TIME` then the user specified timestamp will be recorded or generated if not specified. +If the topic is configured to use `LOG_APPEND_TIME` then the user specified timestamp will be ignored and broker will add in the local broker time. + +The `metrics` and `partitionsFor` methods simply delegate to the same methods on the underlying https://kafka.apache.org/0101/javadoc/org/apache/kafka/clients/producer/Producer.html[`Producer`]. +The `execute` method provides direct access to the underlying https://kafka.apache.org/0101/javadoc/org/apache/kafka/clients/producer/Producer.html[`Producer`]. To use the template, configure a producer factory and provide it in the template's constructor: @@ -83,6 +91,7 @@ header: - `KafkaHeaders.TOPIC` - `KafkaHeaders.PARTITION_ID` - `KafkaHeaders.MESSAGE_KEY` +- `KafkaHeaders.TIMESTAMP` with the message payload being the data. @@ -308,6 +317,7 @@ public interface Acknowledgment { This gives the listener control over when offsets are committed. +[[kafka-listener-annotation]] ===== @KafkaListener Annotation The `@KafkaListener` annotation provides a mechanism for simple POJO listeners: @@ -391,7 +401,15 @@ public void listen(String data, Acknowledgment ack) { } ---- -Finally, metadata about the message is available from message headers: +Finally, metadata about the message is available from message headers, the following header names can be used for retrieving the headers of the message: + +- `KafkaHeaders.RECEIVED_MESSAGE_KEY` +- `KafkaHeaders.RECEIVED_TOPIC` +- `KafkaHeaders.RECEIVED_PARTITION_ID` +- `KafkaHeaders.RECEIVED_MESSAGE_KEY` +- `KafkaHeaders.RECEIVED_TIMESTAMP` +- `KafkaHeaders.TIMESTAMP_TYPE` + [source, java] ---- @@ -399,7 +417,9 @@ Finally, metadata about the message is available from message headers: public void listen(@Payload String foo, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition, - @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { + @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, + @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts + ) { ... } ---- diff --git a/src/reference/asciidoc/testing.adoc b/src/reference/asciidoc/testing.adoc index eb8d1891..833762cb 100644 --- a/src/reference/asciidoc/testing.adoc +++ b/src/reference/asciidoc/testing.adoc @@ -123,6 +123,27 @@ public static Matcher> hasValue(V value) { ... } * @return a Matcher that matches the partition in a consumer record. */ public static Matcher> hasPartition(int partition) { ... } + +/** + * Matcher testing the timestamp of a {@link ConsumerRecord} asssuming the topic has been set with + * {@link org.apache.kafka.common.record.TimestampType#CREATE_TIME CreateTime}. + * + * @param ts timestamp of the consumer record. + * @return a Matcher that matches the timestamp in a consumer record. + */ +public static Matcher> hasTimestamp(long ts) { + return hasTimestamp(TimestampType.CREATE_TIME, ts); +} + +/** + * Matcher testing the timestamp of a {@link ConsumerRecord} + * @param type timestamp type of the record + * @param ts timestamp of the consumer record. + * @return a Matcher that matches the timestamp in a consumer record. + */ +public static Matcher> hasTimestamp(TimestampType type, long ts) { + return new ConsumerRecordTimestampMatcher(type, ts); +} ---- ==== AssertJ Conditions @@ -148,6 +169,23 @@ public static Condition> value(V value) { ... } * @return a Condition that matches the partition in a consumer record. */ public static Condition> partition(int partition) { ... } + +/** + * @param value the timestamp. + * @return a Condition that matches the timestamp value in a consumer record. + */ +public static Condition> timestamp(long value) { + return new ConsumerRecordTimestampCondition(TimestampType.CREATE_TIME, value); +} + +/** + * @param type the type of timestamp + * @param value the timestamp. + * @return a Condition that matches the timestamp value in a consumer record. + */ +public static Condition> timestamp(TimestampType type, long value) { + return new ConsumerRecordTimestampCondition(type, value); +} ---- ==== Example diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 2652ffb4..dc4650ad 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -1,28 +1,8 @@ -=== What's new in 1.2 Since 1.1 +=== What's new in 1.3 Since 1.2 -This version uses the 0.10.2.x client. +==== Support for Kafka timestamps -=== What's new in 1.1 Since 1.0 - -==== Kafka Client - -This version uses the Apache Kafka 0.10.0.x or 0.10.1.x client. - -==== Batch Listeners - -Listeners can be configured to receive the entire batch of messages returned by the `consumer.poll()` operation, rather than one at a time. - -==== Null Payloads - -Null payloads are used to "delete" keys when using log compaction. - -==== Initial Offset - -When explicitly assigning partitions, you can now configure the initial offset relative to the current position for the consumer group, rather than absolute or relative to the current end. - -==== Seek - -You can now seek the position of each topic/partition. -This can be used to set the initial position during initialization when group management is in use and Kafka assigns the partitions. -You can also seek when an idle container is detected, or at any arbitrary point in your application's execution. -See <> for more information. +`KafkaTemplate` now supports API to add records with timestamps. +New `KafkaHeaders` have been introduced regarding `timestamp` support. +Also new `KafkaConditions.timestamp()` and `KafkaMatchers.hasTimestamp()` testing utilities have been added. +See <>, <> and <> for more details.