GH-716: Null Payload Doc Improvements

Fixes #716

Improve documentation - `null` is not necessarily just for tombstone records.

(cherry picked from commit 9ac47f6)

Author: garyrussell

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

@@ -256,12 +256,16 @@ public void testSimple() throws Exception {
 				.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
@@ -1103,6 +1107,10 @@ static class Listener implements ConsumerSeekAware {
 
 		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);
@@ -1247,8 +1255,10 @@ public void jsonHeaders(Bar foo) { // should be mapped to Foo via Headers
 
 		@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")

src/reference/asciidoc/kafka.adoc

@@ -1631,11 +1631,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.
@@ -1644,7 +1644,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]
 ----
@@ -1654,7 +1654,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]
 ----
@@ -1679,6 +1679,8 @@ static class MultiListenerBean {
 }
 ----
 
+Note that the argument will be `null` not a `KafkaNull`.
+
 [[annotation-error-handling]]
 ==== Handling Exceptions