Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3953: Addressing PR review comments

Browse Source 
Fixes: https://github.com/spring-projects/spring-kafka/issues/3953

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

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
This commit is contained in:
Soby Chacko
2025-06-16 14:58:13 -04:00
parent 3dc1fc3c1e
commit 2be0fc59f7
3 changed files with 39 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 @@ import org.springframework.util.Assert;
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 class DelegatingByTypeSerializer implements Serializer<Object> {
* 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) {

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

@@ -1,61 +0,0 @@
/*
* Copyright 2021-2025 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.kafka.support.serializer;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.kafka.common.serialization.Serializer;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/**
* @author Mahesh Aravind V
*
*/
class DelegatingByTypeSerializerTests {
@Nested
class AssignableTest {
@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);
}
}
}

27
spring-kafka/src/test/java/org/springframework/kafka/support/serializer/DelegatingSerializationTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.kafka.support.serializer;
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 org.springframework.messaging.MessageHeaders;
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 @@ public class DelegatingSerializationTests {
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()));