GH-2154: Fix Tombstones with Delegating Seriali...

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2022 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.
@@ -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);
 	}
 

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);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2022 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.
@@ -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);