spring-projectsGH-2154: Fix Tombestones with Delegating Serial...

Resolves spring-projects#2154

Check for null and return.

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTopicSerializer.java

@@ -74,6 +74,9 @@ public byte[] serialize(String topic, Object data) {
 	@SuppressWarnings("unchecked")
 	@Override
 	public byte[] serialize(String topic, Headers headers, Object data) {
+		if (data == null) {
+			return null;
+		}
 		return ((Serializer<Object>) findDelegate(topic)).serialize(topic, headers, data);
 	}
 

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java

@@ -83,12 +83,18 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 
 	@Override
 	public byte[] serialize(String topic, Object data) {
+		if (data == null) {
+			return null;
+		}
 		Serializer<Object> delegate = findDelegate(data, this.delegates);
 		return delegate.serialize(topic, data);
 	}
 
 	@Override
 	public byte[] serialize(String topic, Headers headers, Object data) {
+		if (data == null) {
+			return null;
+		}
 		Serializer<Object> delegate = findDelegate(data, this.delegates);
 		return delegate.serialize(topic, headers, data);
 	}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingSerializer.java

@@ -190,6 +190,9 @@ public byte[] serialize(String topic, Object data) {
 
 	@Override
 	public byte[] serialize(String topic, Headers headers, Object data) {
+		if (data == null) {
+			return null;
+		}
 		byte[] value = null;
 		String selectorKey = selectorKey();
 		Header header = headers.lastHeader(selectorKey);