Add @PulsarTypeMapping for default topic/schema

This commit introduces the @PulsarTypeMapping annotation which can be
used on message classes to specify default topic and/or schema info.

Author: alesharik

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

@@ -34,6 +34,16 @@ 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:
+
+[source,java,indent=0,subs="verbatim"]
+----
+@PulsarTopic("baz-topic")
+record Baz(String value) {
+}
+----
+
 === 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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2023 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.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;
+
+/**
+ * Specifies default topic and schema for class.
+ *
+ * @author Aleksei Arsenev
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface PulsarTypeMapping {
+
+	/**
+	 * Default topic for class.
+	 * @return topic
+	 */
+	String topic() default "";
+
+	/**
+	 * Default schema type for class.
+	 * @return schema type
+	 */
+	SchemaType schemaType() default SchemaType.NONE;
+
+	/**
+	 * Message key type (must be specified when schema type is {@code KEY_VALUE})
+	 * @return message key type
+	 */
+	Class<?> messageKeyType() default Void.class;
+
+}

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

@@ -39,8 +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;
 
 /**
  * Default schema resolver capable of handling basic message types.
@@ -138,6 +140,14 @@ public <T> Resolved<Schema<T>> resolveSchema(@Nullable Class<?> messageClass, bo
 	@Nullable
 	protected Schema<?> getCustomSchemaOrMaybeDefault(@Nullable Class<?> messageClass, boolean returnDefault) {
 		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 (schema == null && returnDefault) {
 			if (messageClass != null) {
 				try {

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

@@ -17,11 +17,13 @@
 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;
 
 /**
@@ -35,7 +37,7 @@
  */
 public class DefaultTopicResolver implements TopicResolver {
 
-	private final Map<Class<?>, String> customTopicMappings = new LinkedHashMap<>();
+	private final Map<Class<?>, String> customTopicMappings = new ConcurrentHashMap<>();
 
 	/**
 	 * Adds a custom mapping from message type to topic.
@@ -100,7 +102,19 @@ protected Resolved<String> doResolveTopic(@Nullable String userSpecifiedTopic, @
 		if (messageType == null) {
 			return Resolved.failed("Topic must be specified when the message is null");
 		}
-		String topic = this.customTopicMappings.getOrDefault(messageType, defaultTopicSupplier.get());
+
+		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 (topic == null) {
+			topic = defaultTopicSupplier.get();
+		}
 		return topic == null ? Resolved.failed("Topic must be specified when no default topic is configured")
 				: Resolved.of(topic);
 	}

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

@@ -49,6 +49,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.pulsar.annotation.PulsarTypeMapping;
 import org.springframework.pulsar.listener.Proto;
 import org.springframework.pulsar.listener.Proto.Person;
 
@@ -201,6 +202,11 @@ void customMessageTypes() {
 			assertThat(resolver.resolveSchema(Bar.class, true).orElseThrow()).isEqualTo(Schema.BYTES);
 		}
 
+
+		@Test
+		void annotatedMessageType() {
+			assertThat(resolver.resolveSchema(Zaz.class, false).orElseThrow()).isEqualTo(Schema.STRING);
+		}
 	}
 
 	@Nested
@@ -371,4 +377,8 @@ record Bar<T>(T value) {
 	record Zaa(String value) {
 	}
 
+	@PulsarTypeMapping(schemaType = SchemaType.STRING)
+	record Zaz(String value) {
+	}
+
 }

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

@@ -30,6 +30,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import org.springframework.lang.Nullable;
+import org.springframework.pulsar.annotation.PulsarTypeMapping;
 
 /**
  * Unit tests for {@link DefaultTopicResolver}.
@@ -44,6 +45,8 @@ class DefaultTopicResolverTests {
 
 	private static final String fooTopic = "foo-topic1";
 
+	private static final String bazTopic = "baz-topic1";
+
 	private static final String stringTopic = "string-topic1";
 
 	private DefaultTopicResolver resolver = new DefaultTopicResolver();
@@ -111,6 +114,8 @@ static Stream<Arguments> resolveByMessageTypeProvider() {
 				arguments("complexMessageWithUserTopic", userTopic, Foo.class, defaultTopic, userTopic),
 				arguments("complexMessageNoUserTopic", null, Foo.class, defaultTopic, fooTopic),
 				arguments("nullMessageWithUserTopicAndDefault", userTopic, null, defaultTopic, userTopic),
+				arguments("annotationMessageWithUserTopic", userTopic, Baz.class, defaultTopic, userTopic),
+				arguments("annotationMessageNoUserTopic", null, Baz.class, defaultTopic, bazTopic),
 				arguments("nullMessageWithDefault", null, null, defaultTopic, null),
 				arguments("noMatchWithUserTopicAndDefault", userTopic, Bar.class, defaultTopic, userTopic),
 				arguments("noMatchWithUserTopic", userTopic, Bar.class, null, userTopic),
@@ -166,4 +171,8 @@ record Foo(String value) {
 	record Bar(String value) {
 	}
 
+	@PulsarTypeMapping(topic = bazTopic)
+	record Baz(String value) {
+	}
+
 }