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

Support for timestamps in Kafka Template

Browse Source 
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
This commit is contained in:
Biju Kunjummen
2017-02-21 09:24:52 -08:00
committed by Gary Russell
parent 5eec44501d
commit 30b387cc78
19 changed files with 514 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
gradle/wrapper/gradle-wrapper.jar vendored
View File

Binary file not shown.

2
gradle/wrapper/gradle-wrapper.properties vendored
View File

@@ -1,4 +1,4 @@
#Wed Jan 04 10:29:38 EST 2017
#Wed Feb 22 12:35:56 EST 2017
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
zipStoreBase=GRADLE_USER_HOME

43
spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.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.
@@ -17,6 +17,7 @@
package org.springframework.kafka.test.assertj;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.record.TimestampType;
import org.assertj.core.api.Condition;
/**
@@ -24,6 +25,7 @@ import org.assertj.core.api.Condition;
*
* @author Artem Bilan
* @author Gary Russell
* @author Biju Kunjummen
*/
public final class KafkaConditions {
@@ -49,6 +51,25 @@ public final class KafkaConditions {
return new ConsumerRecordValueCondition<>(value);
}
/**
* @param value the timestamp.
* @return a Condition that matches the timestamp value in a consumer record.
* @since 2.0
*/
public static Condition<ConsumerRecord<?, ?>> 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<ConsumerRecord<?, ?>> 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<ConsumerRecord<?, ?>> {
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<ConsumerRecord<?, ?>> {
private final int partition;

73
spring-kafka-test/src/main/java/org/springframework/kafka/test/hamcrest/KafkaMatchers.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.
@@ -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<ConsumerRecord<?, ?>> hasTimestamp(long ts) {
return hasTimestamp(TimestampType.CREATE_TIME, ts);
}
public static class ConsumerRecordKeyMatcher<K> extends DiagnosingMatcher<ConsumerRecord<K, ?>> {
/**
* 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<ConsumerRecord<?, ?>> hasTimestamp(TimestampType type, long ts) {
return new ConsumerRecordTimestampMatcher(type, ts);
}
public static class ConsumerRecordKeyMatcher<K>
extends DiagnosingMatcher<ConsumerRecord<K, ?>> {
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<K, Object> record = (ConsumerRecord<K, Object>) 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<ConsumerRecord<?, ?>> {
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<Object, Object> record = (ConsumerRecord<Object, Object>) 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);
}
}
}

63
spring-kafka-test/src/test/java/org/springframework/kafka/test/hamcrest/KafkaMatchersTests.java Normal file
View File

@@ -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<String, String> 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<String, String> 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: <CreateTime> and value: <123L>"));
assertThat(record, hasTimestamp(123L));
}
}

47
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.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.
@@ -34,8 +34,17 @@ import org.springframework.util.concurrent.ListenableFuture;
* @param <K> the key type.
* @param <V> 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<K, V> {
@@ -61,7 +70,18 @@ public interface KafkaOperations<K, V> {
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
ListenableFuture<SendResult<K, V>> sendDefault(int partition, K key, V data);
ListenableFuture<SendResult<K, V>> 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<SendResult<K, V>> 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<K, V> {
*/
ListenableFuture<SendResult<K, V>> 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<SendResult<K, V>> 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<K, V> {
* @param data the data.
* @return a Future for the {@link SendResult}.
*/
ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data);
ListenableFuture<SendResult<K, V>> 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<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
/**
* Send a message with routing information in message headers. The message payload

16
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
View File

@@ -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<K, V> implements KafkaOperations<K, V> {
@@ -141,10 +142,15 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
}
@Override
public ListenableFuture<SendResult<K, V>> sendDefault(int partition, K key, V data) {
public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data) {
return send(this.defaultTopic, partition, key, data);
}
@Override
public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data) {
return send(this.defaultTopic, partition, timestamp, key, data);
}
@Override
public ListenableFuture<SendResult<K, V>> send(String topic, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
@@ -158,14 +164,14 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
}
@Override
public ListenableFuture<SendResult<K, V>> send(String topic, int partition, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, partition, null, data);
public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
return doSend(producerRecord);
}
@Override
public ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, timestamp, key, data);
return doSend(producerRecord);
}

18
spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java
View File

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

9
spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.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.
@@ -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<String> topics = new ArrayList<>();
List<Integer> partitions = new ArrayList<>();
List<Long> offsets = new ArrayList<>();
List<String> timestampTypes = new ArrayList<>();
List<Long> 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);
}

8
spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.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.
@@ -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);
}
/**

93
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.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.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<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(
consumerProps);
DefaultKafkaConsumerFactory<Integer, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> 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<Integer, String> 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<MetricName, ? extends Metric> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic(INT_KEY_TOPIC);
template.sendDefault(0, 1487694048607L, null, "foo-ts1");
ConsumerRecord<Integer, String> 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<Integer, String> r2 = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
assertThat(r2).has(value("foo-ts2"));
assertThat(r2).has(timestamp(1487694048610L));
Map<MetricName, ? extends Metric> metrics = template.execute(Producer::metrics);
assertThat(metrics).isNotNull();
metrics = template.metrics();
assertThat(metrics).isNotNull();
List<PartitionInfo> partitions = template.partitionsFor(INT_KEY_TOPIC);
assertThat(partitions).isNotNull();
assertThat(partitions.size()).isEqualTo(2);
pf.destroy();
}
@Test
public void testWithMessage() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
Message<String> message1 = MessageBuilder.withPayload("foo-message")
.setHeader(KafkaHeaders.TOPIC, INT_KEY_TOPIC)
.setHeader(KafkaHeaders.PARTITION_ID, 0)
.build();
template.send(message1);
ConsumerRecord<Integer, String> r1 = KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
assertThat(r1).has(value("foo-message"));
Message<String> 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<Integer, String> 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<String, Object> 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.<ConsumerRecord<String, String>>allOf(key("foo"), value("bar")));
consumer.close();
pf.createProducer().close();
pf.destroy();
}
}

71
spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java Normal file
View File

@@ -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<ConsumerRecord<?, ?>> 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));
}
}

1
src/checkstyle/checkstyle-suppressions.xml
View File

@@ -6,4 +6,5 @@
<suppress files="package-info\.java" checks=".*" />
<suppress files="[\\/]test[\\/]" checks="RequireThis" />
<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
<suppress files="KafkaMatchersTests" checks="RegexpSinglelineJava" />
</suppressions>

3
src/reference/asciidoc/appendix.adoc
View File

@@ -1 +1,4 @@
[[history]]
== Change History
include::./changes-1.0-1.1.adoc[]

25
src/reference/asciidoc/changes-1.0-1.1.adoc Normal file
View File

@@ -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 <<seek>> for more information.

4
src/reference/asciidoc/index-docinfo.xml
View File

@@ -13,6 +13,10 @@
<firstname>Artem</firstname>
<surname>Bilan</surname>
</author>
<author>
<firstname>Biju</firstname>
<surname>Kunjummen</surname>
</author>
</authorgroup>
<legalnotice>
<para>

36
src/reference/asciidoc/kafka.adoc
View File

@@ -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<SendResult<K, V>> sendDefault(V data);
ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(int partition, K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, V data);
ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, int partition, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(Message<?> message);
@@ -45,10 +48,15 @@ interface ProducerCallback<K, V, T> {
----
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
) {
...
}
----

38
src/reference/asciidoc/testing.adoc
View File

@@ -123,6 +123,27 @@ public static <V> Matcher<ConsumerRecord<?, V>> hasValue(V value) { ... }
* @return a Matcher that matches the partition in a consumer record.
*/
public static Matcher<ConsumerRecord<?, ?>> 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<ConsumerRecord<?, ?>> 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<ConsumerRecord<?, ?>> hasTimestamp(TimestampType type, long ts) {
return new ConsumerRecordTimestampMatcher(type, ts);
}
----
==== AssertJ Conditions
@@ -148,6 +169,23 @@ public static <V> Condition<ConsumerRecord<?, V>> value(V value) { ... }
* @return a Condition that matches the partition in a consumer record.
*/
public static Condition<ConsumerRecord<?, ?>> partition(int partition) { ... }
/**
* @param value the timestamp.
* @return a Condition that matches the timestamp value in a consumer record.
*/
public static Condition<ConsumerRecord<?, ?>> 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<ConsumerRecord<?, ?>> timestamp(TimestampType type, long value) {
return new ConsumerRecordTimestampCondition(type, value);
}
----
==== Example

32
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<seek>> 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 <<kafka-template>>, <<kafka-listener-annotation>> and <<testing>> for more details.