GH-3953: Addressing PR review comments

Fixes: #3953

**Auto-cherry-pick to `3.3.x` & `3.2.x`**"

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

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

@@ -16,8 +16,10 @@
 
 package org.springframework.kafka.support.serializer;
 
-import java.util.*;
+import java.util.Comparator;
+import java.util.Map;
 import java.util.Map.Entry;
+import java.util.TreeMap;
 
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.header.Headers;
@@ -39,16 +41,18 @@
 public class DelegatingByTypeSerializer implements Serializer<Object> {
 
 	private static final Comparator<Class<?>> DELEGATES_ASSIGNABILITY_COMPARATOR =
-			(type1, type2) -> {
+			(class1, class2) -> {
 
-				if (type1.isAssignableFrom(type2)) {
-					return 1;
+				if (class1 == class2) {
+					return 0; // Classes are the same
 				}
-				if (type2.isAssignableFrom(type1)) {
-					return -1;
+				if (class1.isAssignableFrom(class2)) {
+					return 1; // class2 is a superclass or superinterface of class1, so class2 should come first
 				}
-
-				return 0;
+				if (class2.isAssignableFrom(class1)) {
+					return -1; // class1 is a superclass or superinterface of class2, so class1 should come first
+				}
+				return class1.getName().compareTo(class2.getName()); // If no inheritance relation, compare by name
 			};
 
 	private final Map<Class<?>, Serializer<?>> delegates = new TreeMap<>(DELEGATES_ASSIGNABILITY_COMPARATOR);
@@ -70,11 +74,9 @@ public DelegatingByTypeSerializer(Map<Class<?>, Serializer<?>> delegates) {
 	 * is assignable from the target object's class. When multiple matches are possible,
 	 * the most specific matching class is selected — that is, the closest match in the
 	 * class hierarchy.
-	 *
 	 * @param delegates the delegates
 	 * @param assignable true if {@link #findDelegate(Object, Map)} should consider assignability to
 	 * the key rather than an exact match.
-	 *
 	 * @since 2.8.3
 	 */
 	public DelegatingByTypeSerializer(Map<Class<?>, Serializer<?>> delegates, boolean assignable) {

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

@@ -19,6 +19,7 @@
 import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 import org.apache.kafka.common.errors.SerializationException;
@@ -41,6 +42,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -122,6 +124,31 @@ void testWithPropertyConfigKeys() {
 		doTestKeys(serializer, deserializer);
 	}
 
+	@Test
+	void shouldOrderDelegatesSoChildComesBeforeParent() {
+		class Parent { }
+
+		class Child extends Parent { }
+
+		Serializer<?> mockParentSerializer = mock(Serializer.class);
+		Serializer<?> mockChildSerializer = mock(Serializer.class);
+
+		// Using LinkedHashMap to ensure the order is always wrong
+		Map<Class<?>, Serializer<?>> delegates = new LinkedHashMap<>();
+		delegates.put(Parent.class, mockParentSerializer);
+		delegates.put(Child.class, mockChildSerializer);
+
+		DelegatingByTypeSerializer serializer = new DelegatingByTypeSerializer(delegates, true);
+
+
+		Serializer<?> childSerializer = serializer.findDelegate(mock(Child.class));
+		Serializer<?> parentSerializer = serializer.findDelegate(mock(Parent.class));
+
+
+		assertThat(childSerializer).isEqualTo(mockChildSerializer);
+		assertThat(parentSerializer).isEqualTo(mockParentSerializer);
+	}
+
 	private void doTest(DelegatingSerializer serializer, DelegatingDeserializer deserializer) {
 		Headers headers = new RecordHeaders();
 		headers.add(new RecordHeader(DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR, "bytes".getBytes()));