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

GH-716: Null Payload Doc Improvements

Browse Source 
Fixes https://github.com/spring-projects/spring-kafka/issues/716

Improve documentation - `null` is not necessarily just for tombstone records.
This commit is contained in:
Gary Russell
2018-06-26 12:07:57 -04:00
committed by Artem Bilan
parent 7d85c56dc7
commit 9ac47f6b18
2 changed files with 18 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -257,12 +257,16 @@ public class EnableKafkaIntegrationTests {
.isEqualTo("clientIdViaAnnotation-0");
template.send("annotated11", 0, "foo");
template.flush();
assertThat(this.listener.latch7.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.consumerRef.get()).isNotNull();
assertThat(this.listener.latch7String).isEqualTo("foo");
assertThat(this.recordFilter.called).isTrue();
template.send("annotated11", 0, null);
assertThat(this.listener.latch7a.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.latch7String).isNull();
MessageListenerContainer rebalanceConcurrentContainer = registry.getListenerContainer("rebalanceListener");
assertThat(rebalanceConcurrentContainer).isNotNull();
MessageListenerContainer rebalanceContainer = (MessageListenerContainer) KafkaTestUtils
@@ -1114,6 +1118,10 @@ public class EnableKafkaIntegrationTests {
private final CountDownLatch latch7 = new CountDownLatch(1);
private final CountDownLatch latch7a = new CountDownLatch(2);
private volatile String latch7String;
private final CountDownLatch latch8 = new CountDownLatch(1);
private final CountDownLatch latch9 = new CountDownLatch(1);
@@ -1262,8 +1270,10 @@ public class EnableKafkaIntegrationTests {
@KafkaListener(id = "rebalanceListener", topics = "annotated11", idIsGroup = false,
containerFactory = "kafkaRebalanceListenerContainerFactory")
public void listen7(String foo) {
public void listen7(@Payload(required = false) String foo) {
this.latch7String = foo;
this.latch7.countDown();
this.latch7a.countDown();
}
@KafkaListener(id = "quux", topics = "annotated12")

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

@@ -1686,11 +1686,11 @@ If the converter has no converter (either because Jackson is not present, or it
IMPORTANT: The Jackson `ObjectMapper` (even if provided) will be enhanced to support deserializing `org.springframework.util.MimeType` objects, often used in the `spring-messaging` `contentType` header.
If you don't wish your mapper to be enhanced in this way, for some reason, you should subclass the `DefaultKafkaHeaderMapper` and override `getObjectMapper()` to return your mapper.
==== Log Compaction
==== Null Payloads and Log Compaction 'Tombstone' Records
When using https://kafka.apache.org/documentation/#compaction[Log Compaction], it is possible to send and receive messages with `null` payloads which identifies the deletion of a key.
Starting with _version 1.0.3_, this is now fully supported.
It is also possible to receive `null` values for other reasons - such as a `Deserializer` that might return `null` when it can't deserialize a value.
To send a `null` payload using the `KafkaTemplate` simply pass null into the value argument of the `send()` methods.
One exception to this is the `send(Message<?> message)` variant.
@@ -1699,7 +1699,7 @@ For convenience, the static `KafkaNull.INSTANCE` is provided.
When using a message listener container, the received `ConsumerRecord` will have a `null` `value()`.
To configure the `@KafkaListener` to handle `null` payloads, you must use the `@Payload` annotation with `required = false`; you will usually also need the key so your application knows which key was "deleted":
To configure the `@KafkaListener` to handle `null` payloads, you must use the `@Payload` annotation with `required = false`; if it's a tombstone message for a compacted log, you will usually also need the key so your application can determine which key was "deleted":
[source, java]
----
@@ -1709,7 +1709,7 @@ public void listen(@Payload(required = false) String value, @Header(KafkaHeaders
}
----
When using a class-level `@KafkaListener`, some additional configuration is needed - a `@KafkaHandler` method with a `KafkaNull` payload:
When using a class-level `@KafkaListener` with multiple `@KafkaHandler` methods, some additional configuration is needed - a `@KafkaHandler` method with a `KafkaNull` payload:
[source, java]
----
@@ -1734,6 +1734,8 @@ static class MultiListenerBean {
}
----
Note that the argument will be `null` not a `KafkaNull`.
[[annotation-error-handling]]
==== Handling Exceptions