Add auto-configuration for PulsarAdministration (#72)

* Auto-configures a `PulsarAdministration`
* Reconciles some properties for admin and regular clients
* Adds docs for config props (see #50)

Resolves #41

buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java

@@ -34,6 +34,7 @@ import org.gradle.api.tasks.TaskAction;
  * @author Andy Wilkinson
  * @author Phillip Webb
  * @author Chris Bono
+ * @author Alexander Preuß
  */
 public class DocumentConfigurationProperties extends DefaultTask {
 
@@ -69,6 +70,7 @@ public class DocumentConfigurationProperties extends DefaultTask {
 			c.accept("spring.pulsar.consumer");
 			c.accept("spring.pulsar.listener");
 		});
+		snippets.add("application-properties.pulsar-administration", "Pulsar Administration Properties", (c) -> c.accept("spring.pulsar.administration"));
 		snippets.writeTo(this.outputDir.toPath());
 	}
 }

spring-pulsar-docs/src/main/asciidoc/application-properties.adoc

@@ -15,3 +15,5 @@ include::application-properties/pulsar-client.adoc[]
 include::application-properties/pulsar-producer.adoc[]
 
 include::application-properties/pulsar-consumer.adoc[]
+
+include::application-properties/pulsar-administration.adoc[]

spring-pulsar-docs/src/main/asciidoc/pulsar.adoc

@@ -571,6 +571,42 @@ ProducerInterceptor secondInterceptor() {
 ----
 ====
 
+[[pulsar-admin]]
+==== Pulsar Admin
+On the Pulsar administration side, Spring Boot auto-configuration provides a `PulsarAdministration` to manage Pulsar clusters.
+The administration implements an interface called `PulsarAdminOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarAdminOperations.html[a 'createOrModify' method] to handle topic administration through its contract.
+
+When using the Pulsar Spring Boot Starter, you get the `PulsarAdministration` auto-configured.
+By default, the application tries to connect to a local Pulsar instance at `http://localhost:8080`. However, there are many application properties available to configure the client.
+
+.[.underline]#Click ##here## to view the available **Pulsar Administration Properties**#.
+[%collapsible]
+====
+include::application-properties/pulsar-administration.adoc[lines=3..-1]
+====
+
+On initialization, the `PulsarAdministration` checks if there are any `PulsarTopic` beans in the application context.
+For all such beans, the `PulsarAdministration` will either create the corresponding topic, or if necessary modify the number of partitions.
+
+Below is an example how to add `PulsarTopic` beans to let the `PulsarAdministration` auto-create topics for you.
+
+====
+[source,java]
+----
+@Bean
+PulsarTopic simpleTopic {
+	// This will create a non-partitioned topic in the public/default namespace
+	return PulsarTopic.builder("simple-topic").build();
+}
+
+@Bean
+PulsarTopic partitionedTopic {
+	// This will create a partitioned topic with 3 partitions in the provided tenant and namespace
+	return PulsarTopic.builder("persistent://my-tenant/my-namespace/partitioned-topic", 3).build();
+}
+----
+====
+
 
 ==== Appendix
 The reference documentation has the following appendices:

spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java

@@ -31,6 +31,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.pulsar.annotation.PulsarListener;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.PulsarTopic;
 
 @SpringBootApplication
 public class SpringPulsarBootApp {
@@ -107,13 +108,36 @@ public class SpringPulsarBootApp {
 		this.logger.info("Message received :" + message);
 	}
 
+	/*
+	 * Create a partitioned topic using PulsarAdministration and then publish to the topic
+	 * and consume from it.
+	 */
+	@Bean
+	PulsarTopic partitionedTopic4() {
+		return PulsarTopic.builder("hello-pulsar-partitioned-4").numberOfPartitions(3).build();
+	}
+
+	@Bean
+	ApplicationRunner runner4(PulsarTemplate<String> pulsarTemplate) {
+		return args -> {
+			for (int i = 0; i < 10; i++) {
+				pulsarTemplate.send("hello-pulsar-partitioned-4", "This is message " + (i + 1));
+			}
+		};
+	}
+
+	@PulsarListener(subscriptionName = "subscription-4", topics = "hello-pulsar-partitioned-4")
+	void listen4(String message) {
+		this.logger.info("Message received from partitioned-topic : " + message);
+	}
+
 	/*
 	 * Publish and then use PulsarListener in batch listening mode.
 	 */
 	@Bean
-	ApplicationRunner runner4(PulsarProducerFactory<Foo> producerFactory) {
+	ApplicationRunner runner5(PulsarProducerFactory<Foo> producerFactory) {
 
-		String topic = "hello-pulsar-exclusive-4";
+		String topic = "hello-pulsar-exclusive-5";
 		PulsarTemplate<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory);
 		pulsarTemplate.setSchema(Schema.JSON(Foo.class));
 		return args -> {
@@ -124,9 +148,9 @@ public class SpringPulsarBootApp {
 		};
 	}
 
-	@PulsarListener(subscriptionName = "subscription-4", topics = "hello-pulsar-exclusive-4",
+	@PulsarListener(subscriptionName = "subscription-5", topics = "hello-pulsar-exclusive-5",
 			schemaType = SchemaType.JSON, batch = true)
-	void listen4(List<Foo> messages) {
+	void listen5(List<Foo> messages) {
 		this.logger.info("records received :" + messages.size());
 		for (Foo message : messages) {
 			this.logger.info("record : " + message);

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java

@@ -33,6 +33,7 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
@@ -98,4 +99,10 @@ public class PulsarAutoConfiguration {
 		return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties());
 	}
 
+	@Bean
+	@ConditionalOnMissingBean(PulsarAdministration.class)
+	public PulsarAdministration pulsarAdministration() {
+		return new PulsarAdministration(this.properties.buildAdminProperties());
+	}
+
 }

spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java

@@ -43,6 +43,7 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
+import org.springframework.pulsar.core.PulsarAdministration;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
@@ -90,7 +91,7 @@ class PulsarAutoConfigurationTests {
 						.hasSingleBean(PulsarTemplate.class).hasSingleBean(PulsarConsumerFactory.class)
 						.hasSingleBean(ConcurrentPulsarListenerContainerFactory.class)
 						.hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class)
-						.hasSingleBean(PulsarListenerEndpointRegistry.class));
+						.hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(PulsarAdministration.class));
 	}
 
 	@Test
@@ -161,6 +162,15 @@ class PulsarAutoConfigurationTests {
 						.isSameAs(listenerAnnotationBeanPostProcessor));
 	}
 
+	@Test
+	void customPulsarAdministrationIsRespected() {
+		PulsarAdministration pulsarAdministration = mock(PulsarAdministration.class);
+		this.contextRunner
+				.withBean("customPulsarAdministration", PulsarAdministration.class, () -> pulsarAdministration)
+				.run((context) -> assertThat(context).hasNotFailed().getBean(PulsarAdministration.class)
+						.isSameAs(pulsarAdministration));
+	}
+
 	@Test
 	void customProducerInterceptorIsUsedInPulsarTemplate() {
 		ProducerInterceptor interceptor = mock(ProducerInterceptor.class);

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

@@ -104,7 +104,8 @@ public class PulsarAdministration
 	}
 
 	private List<String> getMatchingTopicPartitions(PulsarTopic topic, List<String> existingTopics) {
-		return existingTopics.stream().filter(existing -> existing.startsWith(topic + "-partition-")).toList();
+		return existingTopics.stream()
+				.filter(existing -> existing.startsWith(topic.getFullyQualifiedTopicName() + "-partition-")).toList();
 	}
 
 	private void createOrModifyTopicsIfNeeded(Collection<PulsarTopic> topics) {
@@ -134,25 +135,26 @@ public class PulsarAdministration
 					if (topic.isPartitioned()) {
 						List<String> matchingPartitions = getMatchingTopicPartitions(topic, existingTopicsInNamespace);
 						if (matchingPartitions.isEmpty()) {
-							this.logger.debug(() -> "Topic " + topic + " does not exist.");
+							this.logger.debug(() -> "Topic " + topic.getFullyQualifiedTopicName() + " does not exist.");
 							topicsToCreate.add(topic);
 						}
 						else {
 							int numberOfExistingPartitions = matchingPartitions.size();
 							if (numberOfExistingPartitions < topic.numberOfPartitions()) {
-								this.logger.debug(() -> "Topic " + topic + " found with " + numberOfExistingPartitions
-										+ " partitions.");
+								this.logger.debug(() -> "Topic " + topic.getFullyQualifiedTopicName() + " found with "
+										+ numberOfExistingPartitions + " partitions.");
 								topicsToModify.add(topic);
 							}
 							else if (numberOfExistingPartitions > topic.numberOfPartitions()) {
-								throw new IllegalStateException("Topic " + topic + " found with "
-										+ numberOfExistingPartitions + " partitions. Needs to be deleted first.");
+								throw new IllegalStateException("Topic " + topic.getFullyQualifiedTopicName()
+										+ " found with " + numberOfExistingPartitions
+										+ " partitions. Needs to be deleted first.");
 							}
 						}
 					}
 					else {
-						if (!existingTopicsInNamespace.contains(topic.toString())) {
-							this.logger.debug(() -> "Topic " + topic + " does not exist.");
+						if (!existingTopicsInNamespace.contains(topic.getFullyQualifiedTopicName())) {
+							this.logger.debug(() -> "Topic " + topic.getFullyQualifiedTopicName() + " does not exist.");
 							topicsToCreate.add(topic);
 						}
 					}
@@ -168,8 +170,8 @@ public class PulsarAdministration
 	}
 
 	private void createTopics(PulsarAdmin admin, Set<PulsarTopic> topicsToCreate) throws PulsarAdminException {
-		this.logger.debug(() -> "Creating topics: "
-				+ topicsToCreate.stream().map(PulsarTopic::toString).collect(Collectors.joining(",")));
+		this.logger.debug(() -> "Creating topics: " + topicsToCreate.stream()
+				.map(PulsarTopic::getFullyQualifiedTopicName).collect(Collectors.joining(",")));
 		for (PulsarTopic topic : topicsToCreate) {
 			if (topic.isPartitioned()) {
 				admin.topics().createPartitionedTopic(topic.topicName(), topic.numberOfPartitions());
@@ -181,8 +183,8 @@ public class PulsarAdministration
 	}
 
 	private void modifyTopics(PulsarAdmin admin, Set<PulsarTopic> topicsToModify) throws PulsarAdminException {
-		this.logger.debug(() -> "Modifying topics: "
-				+ topicsToModify.stream().map(PulsarTopic::toString).collect(Collectors.joining(",")));
+		this.logger.debug(() -> "Modifying topics: " + topicsToModify.stream()
+				.map(PulsarTopic::getFullyQualifiedTopicName).collect(Collectors.joining(",")));
 		for (PulsarTopic topic : topicsToModify) {
 			admin.topics().updatePartitionedTopic(topic.topicName(), topic.numberOfPartitions());
 		}

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

@@ -69,8 +69,7 @@ public record PulsarTopic(String topicName, int numberOfPartitions) {
 	 * Get the fully-qualified name of the topic.
 	 * @return the fully-qualified topic name
 	 */
-	@Override
-	public String toString() {
+	public String getFullyQualifiedTopicName() {
 		TopicComponents components = this.getComponents();
 		return components.domain + "://" + components.tenant + "/" + components.namespace + "/" + components.name;
 	}

spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java

@@ -48,7 +48,7 @@ public class PulsarContainerProperties {
 		 */
 		BATCH,
 		/**
-		 * Recod ack mode.
+		 * Record ack mode.
 		 */
 		RECORD,
 		/**

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

@@ -57,11 +57,11 @@ public class PulsarAdministrationTests extends AbstractContainerBaseTests {
 		List<String> expectedTopics = expected.stream().<String>mapMulti((topic, consumer) -> {
 			if (topic.isPartitioned()) {
 				for (int i = 0; i < topic.numberOfPartitions(); i++) {
-					consumer.accept(topic + "-partition-" + i);
+					consumer.accept(topic.getFullyQualifiedTopicName() + "-partition-" + i);
 				}
 			}
 			else {
-				consumer.accept(topic.toString());
+				consumer.accept(topic.getFullyQualifiedTopicName());
 			}
 
 		}).toList();