GH-4442: Fix value serializer mappings for different classloaders

Fixes: #4442

When the type mapper is initialized under one classloader and messages
are produced under another (e.g. Spring DevTools restart classloader),
the reverse lookup map keyed on `Class<?>` identity misses the entry
because two `Class<?>` objects representing the same class loaded by
different classloaders are not equal in a `HashMap` lookup. The fallback
writes the FQCN header instead of the configured alias, silently
breaking consumers that depend on the alias.

Fix by changing the reverse map key from `Class<?>` to `String`
(using `clazz.getName()`), which is classloader-agnostic.

The same change is applied to the deprecated `AbstractJavaTypeMapper` for Jackson 2.

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

**Auto-cherry-pick to `4.0.x`**

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/support/mapping/AbstractJavaTypeMapper.java

@@ -79,7 +79,7 @@ public abstract class AbstractJavaTypeMapper implements BeanClassLoaderAware {
 
 	private final Map<String, Class<?>> idClassMapping = new ConcurrentHashMap<String, Class<?>>();
 
-	private final Map<Class<?>, byte[]> classIdMapping = new ConcurrentHashMap<Class<?>, byte[]>();
+	private final Map<String, byte[]> classIdMapping = new ConcurrentHashMap<String, byte[]>();
 
 	private String classIdFieldName = DEFAULT_CLASSID_FIELD_NAME;
 
@@ -143,8 +143,9 @@ public void setBeanClassLoader(ClassLoader classLoader) {
 	}
 
 	protected void addHeader(Headers headers, String headerName, Class<?> clazz) {
-		if (this.classIdMapping.containsKey(clazz)) {
-			headers.add(new RecordHeader(headerName, this.classIdMapping.get(clazz)));
+		byte[] alias = this.classIdMapping.get(clazz.getName());
+		if (alias != null) {
+			headers.add(new RecordHeader(headerName, alias));
 		}
 		else {
 			headers.add(new RecordHeader(headerName, clazz.getName().getBytes(StandardCharsets.UTF_8)));
@@ -177,7 +178,7 @@ private void createReverseMap() {
 		for (Map.Entry<String, Class<?>> entry : this.idClassMapping.entrySet()) {
 			String id = entry.getKey();
 			Class<?> clazz = entry.getValue();
-			this.classIdMapping.put(clazz, id.getBytes(StandardCharsets.UTF_8));
+			this.classIdMapping.put(clazz.getName(), id.getBytes(StandardCharsets.UTF_8));
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/mapping/DefaultJacksonJavaTypeMapper.java

@@ -89,7 +89,7 @@ public class DefaultJacksonJavaTypeMapper implements JacksonJavaTypeMapper, Bean
 
 	private final Map<String, Class<?>> idClassMapping = new ConcurrentHashMap<String, Class<?>>();
 
-	private final Map<Class<?>, byte[]> classIdMapping = new ConcurrentHashMap<Class<?>, byte[]>();
+	private final Map<String, byte[]> classIdMapping = new ConcurrentHashMap<String, byte[]>();
 
 	private String classIdFieldName = DEFAULT_CLASSID_FIELD_NAME;
 
@@ -153,8 +153,9 @@ public void setBeanClassLoader(ClassLoader classLoader) {
 	}
 
 	protected void addHeader(Headers headers, String headerName, Class<?> clazz) {
-		if (this.classIdMapping.containsKey(clazz)) {
-			headers.add(new RecordHeader(headerName, this.classIdMapping.get(clazz)));
+		byte[] alias = this.classIdMapping.get(clazz.getName());
+		if (alias != null) {
+			headers.add(new RecordHeader(headerName, alias));
 		}
 		else {
 			headers.add(new RecordHeader(headerName, clazz.getName().getBytes(StandardCharsets.UTF_8)));
@@ -187,7 +188,7 @@ private void createReverseMap() {
 		for (Map.Entry<String, Class<?>> entry : this.idClassMapping.entrySet()) {
 			String id = entry.getKey();
 			Class<?> clazz = entry.getValue();
-			this.classIdMapping.put(clazz, id.getBytes(StandardCharsets.UTF_8));
+			this.classIdMapping.put(clazz.getName(), id.getBytes(StandardCharsets.UTF_8));
 		}
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/JsonSerializationTests.java

@@ -17,6 +17,9 @@
 package org.springframework.kafka.support.serializer;
 
 import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -60,6 +63,7 @@
  * @author Torsten Schleede
  * @author Gary Russell
  * @author Ivan Ponomarev
+ * @author Soby Chacko
  */
 public class JsonSerializationTests {
 
@@ -434,6 +438,24 @@ void configRejectedIgnoredAfterPropertiesSet() {
 		assertThatIllegalStateException().isThrownBy(() -> ser.configure(configs2, false));
 	}
 
+	@Test
+	void typeMappingHonoredWhenClassLoadedByDifferentClassLoader() throws Exception {
+		DefaultJacksonJavaTypeMapper mapper = new DefaultJacksonJavaTypeMapper();
+		mapper.setIdClassMapping(Map.of("my-alias", Foo.class));
+
+		URL codeSourceUrl = Foo.class.getProtectionDomain().getCodeSource().getLocation();
+		try (URLClassLoader isolatedLoader = new URLClassLoader(new URL[] { codeSourceUrl }, null)) {
+			Class<?> reloadedClass = isolatedLoader.loadClass(Foo.class.getName());
+			assertThat(reloadedClass).isNotSameAs(Foo.class);
+
+			Headers headers = new RecordHeaders();
+			mapper.fromClass(reloadedClass, headers);
+
+			String typeHeader = new String(headers.lastHeader("__TypeId__").value(), StandardCharsets.UTF_8);
+			assertThat(typeHeader).isEqualTo("my-alias");
+		}
+	}
+
 	public static JavaType fooBarJavaType(byte[] data, Headers headers) {
 		if (data[0] == '{' && data[1] == 'f') {
 			return TypeFactory.createDefaultInstance().constructType(Foo.class);