Polish "Add @PulsarTypeMapping for default topi.."

* Introduce a PulsarTypeMappingRegistry to cache annotations
* Expand on the topic/schema resolver tests
* Add support for KEY_VALUE on annotation

Author: onobc

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/schema-info/custom-schema-mapping.adoc

@@ -1,3 +1,4 @@
+==== Configuration properties
 Schema mappings can be configured with the `spring.pulsar.defaults.type-mappings` property.
 The following example uses `application.yml` to add mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively:
 
@@ -17,6 +18,7 @@ spring:
 
 NOTE: The `message-type` is the fully-qualified name of the message class.
 
+==== Schema resolver customizer
 The preferred method of adding mappings is via the property mentioned above.
 However, if more control is needed you can provide a schema resolver customizer to add the mapping(s).
 
@@ -32,3 +34,18 @@ public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer(
 	}
 }
 ----
+
+==== Type mapping annotation
+Another option for specifying default schema information to use for a particular message type is to mark the message class with the `@PulsarTypeMapping` annotation.
+The schema info can be specified via the `schemaType` attribute on the annotation.
+
+The following example configures the system to use JSON as the default schema when producing or consuming messages of type `Foo`:
+
+[source,java,indent=0,subs="verbatim"]
+----
+@PulsarTypeMapping(schemaType = SchemaType.JSON)
+record Foo(String value) {
+}
+----
+
+NOTE: The annotations are looked up on-demand and their result is cached. However, there is still a small performance hit on the first lookup. If you want to disable this feature you can invoke the `usePulsarTypeMappingAnnotations(false)` method on the `DefaultSchemaResolver`.

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc

@@ -35,15 +35,21 @@ NOTE: The `message-type` is the fully-qualified name of the message class.
 WARNING: If the message (or the first message of a `Publisher` input) is `null`, the framework won't be able to determine the topic from it. Another method shall be used to specify the topic if your application is likely to send `null` messages.
 
 === Specified via annotation
-When no topic passed into API and no mappings configured, the system looks for `PulsarTopic` annotation. The following example configures topic for `Baz` class using annotation:
+
+When no topic is passed into the API and there are no custom topic mappings configured, the system looks for a `@PulsarTypeMapping` annotation on the class of the message being produced or consumed.
+The default topic can be specified via the `topic` attribute on the annotation.
+
+The following example configures the default topic to use when producing or consuming messages of type `Foo`:
 
 [source,java,indent=0,subs="verbatim"]
 ----
-@PulsarTopic("baz-topic")
-record Baz(String value) {
+@PulsarTypeMapping(topic = "foo-topic")
+record Foo(String value) {
 }
 ----
 
+NOTE: The annotations are looked up on-demand and their result is cached. However, there is still a small performance hit on the first lookup. If you want to disable this feature you can invoke the `usePulsarTypeMappingAnnotations(false)` method on the `DefaultTopicResolver`.
+
 === Custom topic resolver
 The preferred method of adding mappings is via the property mentioned above.
 However, if more control is needed you can replace the default resolver by proving your own implementation, for example:

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarTypeMapping.java

@@ -16,40 +16,65 @@
 
 package org.springframework.pulsar.annotation;
 
-import org.apache.pulsar.common.schema.SchemaType;
-
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.apache.pulsar.common.schema.SchemaType;
+
 /**
- * Specifies default topic and schema for class.
+ * Specifies default topic and schema info for a message class.
+ * <p>
+ * When a message class is marked with this annotation, the topic/schema resolution
+ * process will use the specified information to determine a topic/schema to use for the
+ * message in process.
  *
  * @author Aleksei Arsenev
+ * @author Chris Bono
  */
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 public @interface PulsarTypeMapping {
 
 	/**
-	 * Default topic for class.
-	 * @return topic
+	 * Default topic for the annotated message class.
+	 * @return default topic for the annotated message class or empty string to indicate
+	 * no default topic is specified
 	 */
 	String topic() default "";
 
 	/**
-	 * Default schema type for class.
-	 * @return schema type
+	 * Default schema type to use for the annotated message class.
+	 * <p>
+	 * Note that when this is set to {@code KEY_VALUE} you must specify the actual key and
+	 * value information via the {@link #messageKeyType()} and
+	 * {@link #messageValueSchemaType()} attributes, respectively.
+	 * @return schema type to use for the annotated message class or {@code NONE} to
+	 * indicate no default schema is specified
 	 */
 	SchemaType schemaType() default SchemaType.NONE;
 
 	/**
-	 * Message key type (must be specified when schema type is {@code KEY_VALUE})
-	 * @return message key type
+	 * The message key type when schema type is set to {@code KEY_VALUE}.
+	 * <p>
+	 * When the {@link #schemaType()} is not set to {@code KEY_VALUE} this attribute is
+	 * ignored.
+	 * @return message key type when using {@code KEY_VALUE} schema type
 	 */
 	Class<?> messageKeyType() default Void.class;
 
+	/**
+	 * The default schema type to use for the value schema when {@link #schemaType()} is
+	 * set to {@code KEY_VALUE}.
+	 * <p>
+	 * When the {@link #schemaType()} is not set to {@code KEY_VALUE} this attribute is
+	 * ignored and the default schema type must be specified via the {@code schemaType}
+	 * attribute.
+	 * @return message value schema type when using {@code KEY_VALUE} schema type
+	 */
+	SchemaType messageValueSchemaType() default SchemaType.NONE;
+
 }

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

@@ -39,10 +39,10 @@
 import org.apache.pulsar.common.schema.SchemaType;
 
 import org.springframework.core.ResolvableType;
-import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.annotation.PulsarTypeMapping;
+import org.springframework.util.Assert;
 
 /**
  * Default schema resolver capable of handling basic message types.
@@ -54,6 +54,7 @@
  * @author Soby Chacko
  * @author Alexander Preuß
  * @author Chris Bono
+ * @author Aleksei Arsenev
  */
 public class DefaultSchemaResolver implements SchemaResolver {
 
@@ -91,6 +92,20 @@ public class DefaultSchemaResolver implements SchemaResolver {
 
 	private final Map<Class<?>, Schema<?>> customSchemaMappings = new LinkedHashMap<>();
 
+	private final PulsarTypeMappingRegistry pulsarTypeMappingRegistry = new PulsarTypeMappingRegistry();
+
+	private boolean usePulsarTypeMappingAnnotations = true;
+
+	/**
+	 * Sets whether to inspect message classes for the
+	 * {@link PulsarTypeMapping @PulsarTypeMapping} annotation during schema resolution.
+	 * @param usePulsarTypeMappingAnnotations whether to inspect messages for the
+	 * annotation
+	 */
+	public void usePulsarTypeMappingAnnotations(boolean usePulsarTypeMappingAnnotations) {
+		this.usePulsarTypeMappingAnnotations = usePulsarTypeMappingAnnotations;
+	}
+
 	/**
 	 * Adds a custom mapping from message type to schema.
 	 * @param messageType the message type
@@ -139,15 +154,18 @@ 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);
-		if (schema == null && messageClass != null) {
-			PulsarTypeMapping annotation = AnnotationUtils.findAnnotation(messageClass, PulsarTypeMapping.class);
-			if (annotation != null && annotation.schemaType() != SchemaType.NONE) {
-				var resolvedSchema = resolveSchema(annotation.schemaType(), messageClass, annotation.messageKeyType());
-				resolvedSchema.ifResolved(objectSchema -> addCustomSchemaMapping(messageClass, objectSchema));
-				schema = resolvedSchema.get().orElse(null);
+
+		// If no custom schema mapping found, look for @PulsarTypeMapping (if enabled)
+		if (this.usePulsarTypeMappingAnnotations && schema == null && messageClass != null) {
+			schema = getAnnotatedSchemaType(messageClass);
+			if (schema != null) {
+				this.addCustomSchemaMapping(messageClass, schema);
 			}
 		}
+
+		// If still no schema, possibly return a default
 		if (schema == null && returnDefault) {
 			if (messageClass != null) {
 				try {
@@ -162,6 +180,30 @@ protected Schema<?> getCustomSchemaOrMaybeDefault(@Nullable Class<?> messageClas
 		return schema;
 	}
 
+	// VisibleForTesting
+	Schema<?> getAnnotatedSchemaType(Class<?> messageClass) {
+		PulsarTypeMapping annotation = this.pulsarTypeMappingRegistry.getTypeMappingFor(messageClass).orElse(null);
+		if (annotation == null || annotation.schemaType() == SchemaType.NONE) {
+			return null;
+		}
+		var schemaType = annotation.schemaType();
+		if (schemaType != SchemaType.KEY_VALUE) {
+			return resolveSchema(annotation.schemaType(), messageClass, null).value().orElse(null);
+		}
+		// handle complicated key value
+		var messageKeyClass = annotation.messageKeyType();
+		Assert.state(messageKeyClass != Void.class,
+				"messageKeyClass can not be Void.class when using KEY_VALUE schema type");
+
+		var messageValueSchemaType = annotation.messageValueSchemaType();
+		Assert.state(messageValueSchemaType != SchemaType.NONE && messageValueSchemaType != SchemaType.KEY_VALUE,
+				() -> "messageValueSchemaType can not be NONE or KEY_VALUE when using KEY_VALUE schema type");
+
+		Schema<?> keySchema = this.resolveSchema(messageKeyClass).orElseThrow();
+		Schema<?> valueSchema = this.resolveSchema(messageValueSchemaType, messageClass, null).orElseThrow();
+		return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE);
+	}
+
 	@Override
 	@SuppressWarnings("unchecked")
 	public <T> Resolved<Schema<T>> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType) {

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

@@ -17,11 +17,10 @@
 package org.springframework.pulsar.core;
 
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 
-import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.annotation.PulsarTypeMapping;
 import org.springframework.util.StringUtils;
@@ -34,10 +33,25 @@
  * {@link #addCustomTopicMapping(Class, String)}.
  *
  * @author Chris Bono
+ * @author Aleksei Arsenev
  */
 public class DefaultTopicResolver implements TopicResolver {
 
-	private final Map<Class<?>, String> customTopicMappings = new ConcurrentHashMap<>();
+	private final Map<Class<?>, String> customTopicMappings = new LinkedHashMap<>();
+
+	private final PulsarTypeMappingRegistry pulsarTypeMappingRegistry = new PulsarTypeMappingRegistry();
+
+	private boolean usePulsarTypeMappingAnnotations = true;
+
+	/**
+	 * Sets whether to inspect message classes for the
+	 * {@link PulsarTypeMapping @PulsarTypeMapping} annotation during topic resolution.
+	 * @param usePulsarTypeMappingAnnotations whether to inspect messages for the
+	 * annotation
+	 */
+	public void usePulsarTypeMappingAnnotations(boolean usePulsarTypeMappingAnnotations) {
+		this.usePulsarTypeMappingAnnotations = usePulsarTypeMappingAnnotations;
+	}
 
 	/**
 	 * Adds a custom mapping from message type to topic.
@@ -102,21 +116,31 @@ protected Resolved<String> doResolveTopic(@Nullable String userSpecifiedTopic, @
 		if (messageType == null) {
 			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.getCustomTopicMappings().get(messageType);
-		if (topic == null) {
-			PulsarTypeMapping annotation = AnnotationUtils.findAnnotation(messageType, PulsarTypeMapping.class);
-			if (annotation != null && !annotation.topic().isBlank()) {
-				this.addCustomTopicMapping(messageType, annotation.topic());
-				topic = annotation.topic();
+		// If no custom topic mapping found, look for @PulsarTypeMapping (if enabled)
+		if (this.usePulsarTypeMappingAnnotations && topic == null) {
+			topic = getAnnotatedTopicInfo(messageType);
+			if (topic != null) {
+				this.addCustomTopicMapping(messageType, topic);
 			}
 		}
 
+		// If still no topic, consult the default topic supplier
 		if (topic == null) {
 			topic = defaultTopicSupplier.get();
 		}
 		return topic == null ? Resolved.failed("Topic must be specified when no default topic is configured")
 				: Resolved.of(topic);
 	}
 
+	// VisibleForTesting
+	String getAnnotatedTopicInfo(Class<?> messageType) {
+		return this.pulsarTypeMappingRegistry.getTypeMappingFor(messageType)
+			.map(PulsarTypeMapping::topic)
+			.filter(StringUtils::hasText)
+			.orElse(null);
+	}
+
 }

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2023-2024 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.pulsar.core;
+
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.pulsar.annotation.PulsarTypeMapping;
+import org.springframework.util.Assert;
+
+/**
+ * A registry that holds the {@link PulsarTypeMapping @PulsarTypeMapping} annotations and
+ * each associated class that is marked with the annotation.
+ * <p>
+ * The annotations are looked up on-demand and the result is cached.
+ * <p>
+ * Once the cache reaches a {@link #maxNumberOfMappingsCached certain size} (default of
+ * {@link #DEFAULT_MAX_CACHE_SIZE}) it is cleared and the annotations will be looked up
+ * again the next time they are requested.
+ *
+ * @author Chris Bono
+ */
+class PulsarTypeMappingRegistry {
+
+	private static final int DEFAULT_MAX_CACHE_SIZE = 1000;
+
+	private final int maxNumberOfMappingsCached;
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private ConcurrentHashMap<Class<?>, Optional<PulsarTypeMapping>> typeMappingsByClass = new ConcurrentHashMap<>();
+
+	PulsarTypeMappingRegistry() {
+		this(DEFAULT_MAX_CACHE_SIZE);
+	}
+
+	PulsarTypeMappingRegistry(int maxNumberOfMappingsCached) {
+		Assert.state(maxNumberOfMappingsCached > 0, "maxNumberOfMappingsCached must be > 0");
+		this.maxNumberOfMappingsCached = maxNumberOfMappingsCached;
+	}
+
+	/**
+	 * Gets the {@link PulsarTypeMapping @PulsarTypeMapping} on the specified class or
+	 * empty if the class is not marked with the annotation.
+	 * @param targetClass the class to check for the annotation
+	 * @return an optional containing the annotation or empty if the class is not marked
+	 * with the annotation.
+	 */
+	Optional<PulsarTypeMapping> getTypeMappingFor(Class<?> targetClass) {
+		var optionalTypeMapping = this.typeMappingsByClass.computeIfAbsent(targetClass, this::findTypeMappingOn);
+		if (this.typeMappingsByClass.size() > this.maxNumberOfMappingsCached) {
+			this.logger
+				.info(() -> "Clearing cache - max entries exceeded (%d)".formatted(this.maxNumberOfMappingsCached));
+			this.typeMappingsByClass = new ConcurrentHashMap<>();
+		}
+		return optionalTypeMapping;
+	}
+
+	// VisibleForTesting
+	protected Optional<PulsarTypeMapping> findTypeMappingOn(Class<?> targetClass) {
+		this.logger.debug(() -> "Looking for @PulsarTypeMapping on " + targetClass);
+		PulsarTypeMapping annotation = AnnotationUtils.findAnnotation(targetClass, PulsarTypeMapping.class);
+		return Optional.ofNullable(annotation);
+	}
+
+}