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

GH-45: Add RECEIVED... Headers

Browse Source 
Resolves #45

Avoid propagation of received headers when sending `o.s.messaging.Message`s.

Also add `Message<?>` methods to the template.

GH-45: Polishing

Remove Extra Spaces
This commit is contained in:
Gary Russell
2016-03-30 17:24:57 -04:00
committed by Artem Bilan
parent dc19f0cd89
commit 753d9bdb89
10 changed files with 146 additions and 30 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.

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

@@ -1,6 +1,6 @@
#Mon Mar 07 20:47:12 EST 2016
#Wed Mar 30 18:31:11 EDT 2016
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-2.11-bin.zip
distributionUrl=https\://services.gradle.org/distributions/gradle-2.12-bin.zip

10
gradlew vendored
View File

@@ -42,11 +42,6 @@ case "`uname`" in
;;
esac
# For Cygwin, ensure paths are in UNIX format before anything is touched.
if $cygwin ; then
[ -n "$JAVA_HOME" ] && JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
fi
# Attempt to set APP_HOME
# Resolve links: $0 may be a link
PRG="$0"
@@ -61,9 +56,9 @@ while [ -h "$PRG" ] ; do
fi
done
SAVED="`pwd`"
cd "`dirname \"$PRG\"`/" >&-
cd "`dirname \"$PRG\"`/" >/dev/null
APP_HOME="`pwd -P`"
cd "$SAVED" >&-
cd "$SAVED" >/dev/null
CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
@@ -114,6 +109,7 @@ fi
if $cygwin ; then
APP_HOME=`cygpath --path --mixed "$APP_HOME"`
CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
JAVACMD=`cygpath --unix "$JAVACMD"`
# We build the pattern for arguments to be converted via cygpath
ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`

26
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java
View File

@@ -21,6 +21,8 @@ import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.messaging.Message;
/**
* The basic Kafka operations contract.
*
@@ -94,10 +96,19 @@ public interface KafkaOperations<K, V> {
*/
Future<RecordMetadata> send(String topic, int partition, K key, V data);
/**
* Send a message with routing information in message headers.
* @param message the message to send.
* @return a Future for the {@link RecordMetadata}.
* @see org.springframework.kafka.support.KafkaHeaders#TOPIC
* @see org.springframework.kafka.support.KafkaHeaders#PARTITION_ID
* @see org.springframework.kafka.support.KafkaHeaders#MESSAGE_KEY
*/
Future<RecordMetadata> send(Message<?> message);
// Sync methods
/**
* Send the data to the default topic with no key or partition;
* wait for result.
@@ -180,6 +191,19 @@ public interface KafkaOperations<K, V> {
RecordMetadata syncSend(String topic, int partition, K key, V data)
throws InterruptedException, ExecutionException;
/**
* Send a message with routing information in message headers.
* @param message the message to send.
* @return a Future for the {@link RecordMetadata}.
* @throws ExecutionException execution exception while awaiting result.
* @throws InterruptedException thread interrupted while awaiting result.
* @see org.springframework.kafka.support.KafkaHeaders#TOPIC
* @see org.springframework.kafka.support.KafkaHeaders#PARTITION_ID
* @see org.springframework.kafka.support.KafkaHeaders#MESSAGE_KEY
*/
RecordMetadata syncSend(Message<?> message)
throws InterruptedException, ExecutionException;
/**
* Flush the producer.
*/

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

@@ -25,9 +25,12 @@ import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.ProducerListenerInvokingCallback;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
/**
@@ -88,28 +91,28 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
}
@Override
public Future<RecordMetadata> send(V data) {
public Future<RecordMetadata> send(V data) {
return send(this.defaultTopic, data);
}
@Override
public Future<RecordMetadata> send(K key, V data) {
public Future<RecordMetadata> send(K key, V data) {
return send(this.defaultTopic, key, data);
}
@Override
public Future<RecordMetadata> send(int partition, K key, V data) {
public Future<RecordMetadata> send(int partition, K key, V data) {
return send(this.defaultTopic, partition, key, data);
}
@Override
public Future<RecordMetadata> send(String topic, V data) {
public Future<RecordMetadata> send(String topic, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
return doSend(producerRecord);
}
@Override
public Future<RecordMetadata> send(String topic, K key, V data) {
public Future<RecordMetadata> send(String topic, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);
return doSend(producerRecord);
}
@@ -121,11 +124,16 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
}
@Override
public Future<RecordMetadata> send(String topic, int partition, K key, V data) {
public Future<RecordMetadata> send(String topic, int partition, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
return doSend(producerRecord);
}
@Override
public Future<RecordMetadata> send(Message<?> message) {
ProducerRecord<K, V> producerRecord = messageToProducerRecord(message);
return doSend(producerRecord);
}
@Override
public RecordMetadata syncSend(V data) throws InterruptedException, ExecutionException {
@@ -180,6 +188,19 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
return future.get();
}
@Override
public RecordMetadata syncSend(Message<?> message)
throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = send(message);
flush();
return future.get();
}
@Override
public void flush() {
this.producer.flush();
}
/**
* Send the producer record.
* @param producerRecord the producer record.
@@ -211,9 +232,14 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
return future;
}
@Override
public void flush() {
this.producer.flush();
@SuppressWarnings({ "rawtypes", "unchecked" })
private ProducerRecord<K, V> messageToProducerRecord(Message<?> message) {
MessageHeaders headers = message.getHeaders();
String topic = headers.get(KafkaHeaders.TOPIC, String.class);
Integer partition = headers.get(KafkaHeaders.PARTITION_ID, Integer.class);
Object key = headers.get(KafkaHeaders.MESSAGE_KEY);
Object payload = message.getPayload();
return new ProducerRecord(topic == null ? this.defaultTopic : topic, partition, key, payload);
}
}

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

@@ -21,24 +21,24 @@ package org.springframework.kafka.support;
*
* @author Artem Bilan
* @author Marius Bogoevici
* @author Gary Russell
*/
public abstract class KafkaHeaders {
private static final String PREFIX = "kafka_";
/**
* The header for topic.
* The header containing the topic when sending data to Kafka.
*/
public static final String TOPIC = PREFIX + "topic";
/**
* The header for message key.
* The header containing the message key when sending data to Kafka.
*/
public static final String MESSAGE_KEY = PREFIX + "messageKey";
/**
* The header for topic partition.
* The header containing the topic partition when sending data to Kafka.
*/
public static final String PARTITION_ID = PREFIX + "partitionId";
@@ -52,4 +52,19 @@ public abstract class KafkaHeaders {
*/
public static final String ACKNOWLEDGMENT = PREFIX + "acknowledgment";
/**
* The header containing the topic from which the message was received.
*/
public static final String RECEIVED_TOPIC = PREFIX + "receivedTopic";
/**
* The header containing the message key for the received message.
*/
public static final String RECEIVED_MESSAGE_KEY = PREFIX + "receivedMessageKey";
/**
* The header containing the topic partition for the received message.
*/
public static final String RECEIVED_PARTITION_ID = PREFIX + "receivedPartitionId";
}

6
spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java
View File

@@ -66,9 +66,9 @@ public class MessagingMessageConverter<K, V> implements MessageConverter<K, V> {
KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId, this.generateTimestamp);
Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
rawHeaders.put(KafkaHeaders.MESSAGE_KEY, record.key());
rawHeaders.put(KafkaHeaders.TOPIC, record.topic());
rawHeaders.put(KafkaHeaders.PARTITION_ID, record.partition());
rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, record.key());
rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, record.topic());
rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, record.partition());
rawHeaders.put(KafkaHeaders.OFFSET, record.offset());
if (acknowledgment != null) {

17
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -83,10 +83,12 @@ public class EnableKafkaIntegrationTests {
assertThat(this.listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
waitListening("bar");
template.send("annotated2", 0, "foo");
template.send("annotated2", 0, 123, "foo");
template.flush();
assertThat(this.listener.latch2.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.key).isEqualTo(123);
assertThat(this.listener.partition).isNotNull();
assertThat(this.listener.topic).isEqualTo("annotated2");
waitListening("baz");
template.send("annotated3", 0, "foo");
@@ -200,14 +202,23 @@ public class EnableKafkaIntegrationTests {
private volatile Acknowledgment ack;
private Integer key;
private String topic;
@KafkaListener(id = "foo", topics = "annotated1")
public void listen1(String foo) {
this.latch1.countDown();
}
@KafkaListener(id = "bar", topicPattern = "annotated2")
public void listen2(@Payload String foo, @Header(KafkaHeaders.PARTITION_ID) int partitionHeader) {
this.partition = partitionHeader;
public void listen2(@Payload String foo,
@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
this.key = key;
this.partition = partition;
this.topic = topic;
this.latch2.countDown();
}

20
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
View File

@@ -35,9 +35,11 @@ import org.junit.Test;
import org.springframework.kafka.listener.ContainerTestUtils;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.ProducerListenerAdapter;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.support.MessageBuilder;
/**
@@ -64,7 +66,6 @@ public class KafkaTemplateTests {
@Override
public void onMessage(ConsumerRecord<Integer, String> record) {
System.out.println(record);
records.add(record);
}
@@ -93,6 +94,23 @@ public class KafkaTemplateTests {
assertThat(received).has(key((Integer) null));
assertThat(received).has(partition(0));
assertThat(received).has(value("qux"));
template.syncSend(MessageBuilder.withPayload("fiz")
.setHeader(KafkaHeaders.TOPIC, TEMPLATE_TOPIC)
.setHeader(KafkaHeaders.PARTITION_ID, 0)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.build());
received = records.poll(10, TimeUnit.SECONDS);
assertThat(received).has(key(2));
assertThat(received).has(partition(0));
assertThat(received).has(value("fiz"));
template.syncSend(MessageBuilder.withPayload("buz")
.setHeader(KafkaHeaders.PARTITION_ID, 0)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.build());
received = records.poll(10, TimeUnit.SECONDS);
assertThat(received).has(key(2));
assertThat(received).has(partition(0));
assertThat(received).has(value("buz"));
}
@Test

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

@@ -24,6 +24,7 @@ Future<RecordMetadata> send(String topic, int partition, V data);
Future<RecordMetadata> send(String topic, int partition, K key, V data);
Future<RecordMetadata> send(Message<?> message);
// Sync methods
@@ -49,6 +50,9 @@ RecordMetadata syncSend(String topic, int partition, V data)
RecordMetadata syncSend(String topic, int partition, K key, V data)
throws InterruptedException, ExecutionException;
RecordMetadata syncSend(Message<?> message)
throws InterruptedException, ExecutionException;
// Flush the producer.
void flush();
@@ -81,6 +85,15 @@ The template can also be configured using standard `<bean/>` definitions.
Then, to use the template, simply invoke one of its methods.
When using the methods with a `Message<?>` parameter, topic, partition and key information is provided in a message
header:
- `KafkaHeaders.TOPIC`
- `KafkaHeaders.PARTITION_ID`
- `KafkaHeaders.MESSAGE_KEY`
with the message payload being the data.
Optionally, you can configure the `KafkaTemplate` with a `ProducerListener` to get an async callback with the
results of the send (success or failure) instead of waiting for the `Future` to complete.
@@ -276,3 +289,16 @@ public void listen(String data, Acknowledgment ack) {
ack.acknowledge();
}
----
Finally, metadata about the message is available from message headers:
[source, java]
----
@KafkaListener(id = "qux", topicPattern = "myTopic1")
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) {
...
}
----