Stop using Class as map key

This commit updates the DefaultTopicResolver and DefaulSchemaResolver to use
String class names rather than Class instances as map keys for their custom
mappings.

Resolves #1078

Author: onobc

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java

@@ -23,12 +23,12 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
-import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.schema.AvroSchema;
@@ -92,7 +92,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
 		BASE_SCHEMA_MAPPINGS.put(LocalTime.class, Schema.LOCAL_TIME);
 	}
 
-	private final Map<Class<?>, Schema<?>> customSchemaMappings = new LinkedHashMap<>();
+	private final Map<String, Schema<?>> customSchemaMappings = new LinkedHashMap<>();
 
 	private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
 
@@ -122,7 +122,7 @@ public void usePulsarMessageAnnotations(boolean usePulsarMessageAnnotations) {
 	 */
 	@Nullable
 	public Schema<?> addCustomSchemaMapping(Class<?> messageType, Schema<?> schema) {
-		return this.customSchemaMappings.put(messageType, schema);
+		return this.customSchemaMappings.put(this.toMessageTypeMapKey(messageType), schema);
 	}
 
 	/**
@@ -133,15 +133,30 @@ public Schema<?> addCustomSchemaMapping(Class<?> messageType, Schema<?> schema)
 	 */
 	@Nullable
 	public Schema<?> removeCustomMapping(Class<?> messageType) {
-		return this.customSchemaMappings.remove(messageType);
+		return this.customSchemaMappings.remove(this.toMessageTypeMapKey(messageType));
 	}
 
 	/**
-	 * Gets the currently registered custom mappings from message type to schema.
-	 * @return unmodifiable map of custom mappings
+	 * Gets the currently registered custom mapping for the specified message type.
+	 * @return optional custom topic registered for the message type
+	 * @deprecated deprecated in favor of {@link #getCustomSchemaMapping(Class)} (Class)}
 	 */
+	@Deprecated(since = "1.2.5", forRemoval = true)
 	public Map<Class<?>, Schema<?>> getCustomSchemaMappings() {
-		return Collections.unmodifiableMap(this.customSchemaMappings);
+		Map<Class<?>, Schema<?>> copyOfMappings = new HashMap<>();
+		this.customSchemaMappings.entrySet()
+			.stream()
+			.map((e) -> copyOfMappings.put(this.fromMessageTypeMapKey(e.getKey()), e.getValue()));
+		return copyOfMappings;
+	}
+
+	/**
+	 * Gets the currently registered custom mapping for the specified message type.
+	 * @param messageType the message type
+	 * @return optional custom topic registered for the message type
+	 */
+	public Optional<Schema<?>> getCustomSchemaMapping(Class<?> messageType) {
+		return Optional.ofNullable(this.customSchemaMappings.get(this.toMessageTypeMapKey(messageType)));
 	}
 
 	@Override
@@ -162,7 +177,7 @@ public <T> Resolved<Schema<T>> resolveSchema(@Nullable Class<?> messageClass, bo
 	@Nullable
 	protected Schema<?> getCustomSchemaOrMaybeDefault(@Nullable Class<?> messageClass, boolean returnDefault) {
 		// Check for custom schema mapping
-		Schema<?> schema = this.customSchemaMappings.get(messageClass);
+		var schema = this.getCustomSchemaMapping(messageClass).orElse(null);
 
 		// If no custom schema mapping found, look for @PulsarMessage (if enabled)
 		if (this.usePulsarMessageAnnotations && schema == null && messageClass != null) {
@@ -289,4 +304,17 @@ private <X> Schema<X> castToType(Schema<?> rawSchema) {
 		return (Schema<X>) rawSchema;
 	}
 
+	private Class<?> fromMessageTypeMapKey(String messageTypeKey) {
+		try {
+			return Class.forName(messageTypeKey);
+		}
+		catch (ClassNotFoundException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	private String toMessageTypeMapKey(Class<?> messageType) {
+		return messageType.getName();
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java

@@ -16,9 +16,10 @@
 
 package org.springframework.pulsar.core;
 
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Supplier;
 
 import org.springframework.beans.BeansException;
@@ -45,7 +46,7 @@ public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
 
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
-	private final Map<Class<?>, String> customTopicMappings = new LinkedHashMap<>();
+	private final Map<String, String> customTopicMappings = new LinkedHashMap<>();
 
 	private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
 
@@ -86,7 +87,7 @@ public void usePulsarMessageAnnotations(boolean usePulsarMessageAnnotations) {
 	 */
 	@Nullable
 	public String addCustomTopicMapping(Class<?> messageType, String topic) {
-		return this.customTopicMappings.put(messageType, topic);
+		return this.customTopicMappings.put(this.toMessageTypeMapKey(messageType), topic);
 	}
 
 	/**
@@ -97,15 +98,31 @@ public String addCustomTopicMapping(Class<?> messageType, String topic) {
 	 */
 	@Nullable
 	public String removeCustomMapping(Class<?> messageType) {
-		return this.customTopicMappings.remove(messageType);
+		return this.customTopicMappings.remove(this.toMessageTypeMapKey(messageType));
 	}
 
 	/**
-	 * Gets the currently registered custom mappings from message type to topic.
+	 * Gets the currently registered custom mappings from message type class name to
+	 * topic.
 	 * @return unmodifiable map of custom mappings
+	 * @deprecated deprecated in favor of {@link #getCustomTopicMapping(Class)}
 	 */
+	@Deprecated(since = "1.2.5", forRemoval = true)
 	public Map<Class<?>, String> getCustomTopicMappings() {
-		return Collections.unmodifiableMap(this.customTopicMappings);
+		Map<Class<?>, String> copyOfMappings = new HashMap<>();
+		this.customTopicMappings.entrySet()
+			.stream()
+			.map((e) -> copyOfMappings.put(this.fromMessageTypeMapKey(e.getKey()), e.getValue()));
+		return copyOfMappings;
+	}
+
+	/**
+	 * Gets the currently registered custom mapping for the specified message type.
+	 * @param messageType the message type
+	 * @return optional custom topic registered for the message type
+	 */
+	public Optional<String> getCustomTopicMapping(Class<?> messageType) {
+		return Optional.ofNullable(this.customTopicMappings.get(this.toMessageTypeMapKey(messageType)));
 	}
 
 	@Override
@@ -141,7 +158,7 @@ protected Resolved<String> doResolveTopic(@Nullable String userSpecifiedTopic, @
 			return Resolved.failed("Topic must be specified when the message is null");
 		}
 		// Check for custom topic mapping
-		String topic = this.customTopicMappings.get(messageType);
+		String topic = this.customTopicMappings.get(this.toMessageTypeMapKey(messageType));
 
 		// If no custom topic mapping found, look for @PulsarMessage (if enabled)
 		if (this.usePulsarMessageAnnotations && topic == null) {
@@ -174,6 +191,19 @@ private String resolveExpression(String v) {
 			.orElseThrow(() -> "Failed to resolve topic expression: %s".formatted(v));
 	}
 
+	private Class<?> fromMessageTypeMapKey(String messageTypeKey) {
+		try {
+			return Class.forName(messageTypeKey);
+		}
+		catch (ClassNotFoundException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	private String toMessageTypeMapKey(Class<?> messageType) {
+		return messageType.getName();
+	}
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		if (beanFactory instanceof ConfigurableBeanFactory configurableBeanFactory) {

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java

@@ -79,6 +79,7 @@ private static String sanitizedClassName(Class<?> clazz) {
 	@Nested
 	class CustomSchemaMappingsAPI {
 
+		@SuppressWarnings("removal")
 		@Test
 		void noMappingsByDefault() {
 			assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP).isEmpty();
@@ -88,14 +89,13 @@ void noMappingsByDefault() {
 		void addMappings() {
 			Schema<?> previouslyMappedSchema = resolver.addCustomSchemaMapping(Foo.class, Schema.STRING);
 			assertThat(previouslyMappedSchema).isNull();
-			assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
-				.containsEntry(Foo.class, Schema.STRING);
+			assertThat(resolver.getCustomSchemaMapping(Foo.class)).hasValue(Schema.STRING);
 			previouslyMappedSchema = resolver.addCustomSchemaMapping(Foo.class, Schema.BOOL);
 			assertThat(previouslyMappedSchema).isEqualTo(Schema.STRING);
-			assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
-				.containsEntry(Foo.class, Schema.BOOL);
+			assertThat(resolver.getCustomSchemaMapping(Foo.class)).hasValue(Schema.BOOL);
 		}
 
+		@SuppressWarnings("removal")
 		@Test
 		void removeMappings() {
 			Schema<?> previouslyMappedSchema = resolver.removeCustomMapping(Foo.class);
@@ -423,10 +423,11 @@ void annotatedMessageType() {
 				.endsWith(JsonMsgType.class.getSimpleName());
 
 			// verify added to custom mappings
-			assertThat(resolver.getCustomSchemaMappings().get(JsonMsgType.class)).isSameAs(resolvedSchema);
+			assertThat(resolver.getCustomSchemaMapping(JsonMsgType.class))
+				.hasValueSatisfying((v) -> assertThat(v).isSameAs(resolvedSchema));
 
 			// verify subsequent calls skip resolution again
-			assertThat(resolver.resolveSchema(JsonMsgType.class, false).orElseThrow()).isSameAs(resolvedSchema);
+			assertThat(resolver.resolveSchema(JsonMsgType.class, false).orElseThrow()).isEqualTo(resolvedSchema);
 			verify(resolver, times(1)).getAnnotatedSchemaType(JsonMsgType.class);
 		}
 

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java

@@ -166,7 +166,7 @@ void annotatedMessageTypeWithTopicInfo() {
 			assertThat(resolver.resolveTopic(null, Baz.class, () -> defaultTopic).value().orElse(null))
 				.isEqualTo(bazTopic);
 			// verify added to custom mappings
-			assertThat(resolver.getCustomTopicMappings().get(Baz.class)).isEqualTo(bazTopic);
+			assertThat(resolver.getCustomTopicMapping(Baz.class)).hasValue(bazTopic);
 			// verify subsequent calls skip resolution again
 			assertThat(resolver.resolveTopic(null, Baz.class, () -> defaultTopic).value().orElse(null))
 				.isEqualTo(bazTopic);
@@ -242,6 +242,7 @@ void resetResolver() {
 			resolver = new DefaultTopicResolver();
 		}
 
+		@SuppressWarnings("removal")
 		@Test
 		void noMappingsByDefault() {
 			resolver = new DefaultTopicResolver();
@@ -254,14 +255,13 @@ void addMappings() {
 			String topic2 = "bar-topic";
 			String previouslyMappedTopic = resolver.addCustomTopicMapping(Foo.class, topic1);
 			assertThat(previouslyMappedTopic).isNull();
-			assertThat(resolver.getCustomTopicMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
-				.containsEntry(Foo.class, topic1);
+			assertThat(resolver.getCustomTopicMapping(Foo.class)).hasValue(topic1);
 			previouslyMappedTopic = resolver.addCustomTopicMapping(Foo.class, topic2);
 			assertThat(previouslyMappedTopic).isEqualTo(topic1);
-			assertThat(resolver.getCustomTopicMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
-				.containsEntry(Foo.class, topic2);
+			assertThat(resolver.getCustomTopicMapping(Foo.class)).hasValue(topic2);
 		}
 
+		@SuppressWarnings("removal")
 		@Test
 		void removeMappings() {
 			String previouslyMappedTopic = resolver.removeCustomMapping(Foo.class);