GH-41: Add auto-created topics

See #41

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

@@ -0,0 +1,192 @@
+/*
+ * Copyright 2022 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.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.commons.logging.LogFactory;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * An administration class that delegates to {@link PulsarAdmin} to create and manage
+ * topics defined in the application context.
+ *
+ * @author Chris Bono
+ * @author Alexander Preuß
+ */
+public class PulsarAdministration
+		implements ApplicationContextAware, SmartInitializingSingleton, PulsarAdministrationOperations {
+
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
+
+	private final PulsarAdminBuilder adminBuilder;
+
+	private ApplicationContext applicationContext;
+
+	/**
+	 * Construct a {@code PulsarAdministration} instance using the given configuration for
+	 * the underlying {@link PulsarAdmin}.
+	 * @param adminConfig the {@link PulsarAdmin} configuration
+	 */
+	public PulsarAdministration(Map<String, Object> adminConfig) {
+		this.adminBuilder = PulsarAdmin.builder().loadConf(adminConfig);
+	}
+
+	/**
+	 * Construct a {@code PulsarAdministration} instance using the given builder for the
+	 * underlying {@link PulsarAdmin}.
+	 * @param adminBuilder the {@link PulsarAdminBuilder}
+	 */
+	public PulsarAdministration(PulsarAdminBuilder adminBuilder) {
+		this.adminBuilder = adminBuilder;
+	}
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		initialize();
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		this.applicationContext = applicationContext;
+	}
+
+	private void initialize() {
+		Collection<PulsarTopic> topics = this.applicationContext.getBeansOfType(PulsarTopic.class, false, false)
+				.values();
+		createOrModifyTopicsIfNeeded(topics);
+	}
+
+	private PulsarAdmin createAdminClient() throws PulsarClientException {
+		return this.adminBuilder.build();
+	}
+
+	@Override
+	public void createOrModifyTopics(PulsarTopic... topics) {
+		createOrModifyTopicsIfNeeded(Arrays.asList(topics));
+	}
+
+	private Map<String, List<PulsarTopic>> getTopicsPerNamespace(Collection<PulsarTopic> topics) {
+		return topics.stream().collect(Collectors.groupingBy(this::getTopicNamespaceIdentifier));
+	}
+
+	private String getTopicNamespaceIdentifier(PulsarTopic topic) {
+		return topic.getComponents().tenant() + "/" + topic.getComponents().namespace();
+	}
+
+	private List<String> getMatchingTopicPartitions(PulsarTopic topic, List<String> existingTopics) {
+		return existingTopics.stream().filter(existing -> existing.startsWith(topic + "-partition-")).toList();
+	}
+
+	private void createOrModifyTopicsIfNeeded(Collection<PulsarTopic> topics) {
+		if (CollectionUtils.isEmpty(topics)) {
+			return;
+		}
+
+		try (PulsarAdmin admin = createAdminClient()) {
+			doCreateOrModifyTopicsIfNeeded(admin, topics);
+		}
+		catch (PulsarClientException e) {
+			throw new IllegalStateException("Could not create PulsarAdmin", e);
+		}
+	}
+
+	private void doCreateOrModifyTopicsIfNeeded(PulsarAdmin admin, Collection<PulsarTopic> topics) {
+		Map<String, List<PulsarTopic>> topicsPerNamespace = getTopicsPerNamespace(topics);
+
+		Set<PulsarTopic> topicsToCreate = new HashSet<>();
+		Set<PulsarTopic> topicsToModify = new HashSet<>();
+
+		topicsPerNamespace.forEach((namespace, requestedTopics) -> {
+			try {
+				List<String> existingTopicsInNamespace = admin.topics().getList(namespace);
+
+				for (PulsarTopic topic : requestedTopics) {
+					if (topic.isPartitioned()) {
+						List<String> matchingPartitions = getMatchingTopicPartitions(topic, existingTopicsInNamespace);
+						if (matchingPartitions.isEmpty()) {
+							this.logger.debug(() -> "Topic " + topic + " does not exist.");
+							topicsToCreate.add(topic);
+						}
+						else {
+							int numberOfExistingPartitions = matchingPartitions.size();
+							if (numberOfExistingPartitions < topic.numberOfPartitions()) {
+								this.logger.debug(() -> "Topic " + topic + " 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.");
+							}
+						}
+					}
+					else {
+						if (!existingTopicsInNamespace.contains(topic.toString())) {
+							this.logger.debug(() -> "Topic " + topic + " does not exist.");
+							topicsToCreate.add(topic);
+						}
+					}
+				}
+
+				createTopics(admin, topicsToCreate);
+				modifyTopics(admin, topicsToModify);
+			}
+			catch (PulsarAdminException e) {
+				throw new RuntimeException(e);
+			}
+		});
+	}
+
+	private void createTopics(PulsarAdmin admin, Set<PulsarTopic> topicsToCreate) throws PulsarAdminException {
+		this.logger.debug(() -> "Creating topics: "
+				+ topicsToCreate.stream().map(PulsarTopic::toString).collect(Collectors.joining(",")));
+		for (PulsarTopic topic : topicsToCreate) {
+			if (topic.isPartitioned()) {
+				admin.topics().createPartitionedTopic(topic.topicName(), topic.numberOfPartitions());
+			}
+			else {
+				admin.topics().createNonPartitionedTopic(topic.topicName());
+			}
+		}
+	}
+
+	private void modifyTopics(PulsarAdmin admin, Set<PulsarTopic> topicsToModify) throws PulsarAdminException {
+		this.logger.debug(() -> "Modifying topics: "
+				+ topicsToModify.stream().map(PulsarTopic::toString).collect(Collectors.joining(",")));
+		for (PulsarTopic topic : topicsToModify) {
+			admin.topics().updatePartitionedTopic(topic.topicName(), topic.numberOfPartitions());
+		}
+	}
+
+}

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2022 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;
+
+/**
+ * The Pulsar administration contract.
+ *
+ * @author Chris Bono
+ * @author Alexander Preuß
+ */
+public interface PulsarAdministrationOperations {
+
+	/**
+	 * Create or modify the given topics.
+	 * @param topics the topics to create or change
+	 */
+	void createOrModifyTopics(PulsarTopic... topics);
+
+}

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

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2022 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 org.apache.pulsar.common.naming.TopicDomain;
+
+/**
+ * Model class for a Pulsar topic.
+ *
+ * Use the {@link PulsarTopicBuilder} to create instances like this:
+ *
+ * <pre>{@code
+ * 	PulsarTopic topic = PulsarTopic.builder("topic-name").build();
+ * }</pre>
+ * @param topicName the topic name
+ * @param numberOfPartitions the number of partitions, or 0 for non-partitioned topics
+ *
+ * @author Chris Bono
+ * @author Alexander Preuß
+ */
+public record PulsarTopic(String topicName, int numberOfPartitions) {
+
+	public static PulsarTopicBuilder builder(String topicName) {
+		return new PulsarTopicBuilder(topicName);
+	}
+
+	/**
+	 * Checks if the topic is partitioned.
+	 * @return true if the topic is partitioned
+	 */
+	public boolean isPartitioned() {
+		return this.numberOfPartitions != 0;
+	}
+
+	/**
+	 * Get the individual identifying components of a Pulsar topic.
+	 * @return {@link TopicComponents}
+	 */
+	public TopicComponents getComponents() {
+		String[] splitTopic = this.topicName().split("/");
+		if (splitTopic.length == 1) { // e.g. 'my-topic'
+			return new TopicComponents(TopicDomain.persistent, "public", "default", splitTopic[0]);
+		}
+		else if (splitTopic.length == 3) { // e.g. 'public/default/my-topic'
+			return new TopicComponents(TopicDomain.persistent, splitTopic[0], splitTopic[1], splitTopic[2]);
+		}
+		else if (splitTopic.length == 5) { // e.g. 'persistent://public/default/my-topic'
+			String type = splitTopic[0].replace(":", "");
+			return new TopicComponents(TopicDomain.getEnum(type), splitTopic[2], splitTopic[3], splitTopic[4]);
+		}
+		throw new IllegalArgumentException("Topic name '" + this + "' has unexpected components.");
+
+	}
+
+	/**
+	 * Get the fully-qualified name of the topic.
+	 * @return the fully-qualified topic name
+	 */
+	@Override
+	public String toString() {
+		TopicComponents components = this.getComponents();
+		return components.domain + "://" + components.tenant + "/" + components.namespace + "/" + components.name;
+	}
+
+	/**
+	 * Model class for the individual identifying components of a Pulsar topic.
+	 * @param domain the topic domain
+	 * @param tenant the topic tenant
+	 * @param namespace the topic namespace
+	 * @param name the topic name
+	 */
+	record TopicComponents(TopicDomain domain, String tenant, String namespace, String name) {
+
+	}
+}

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2022 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;
+
+/**
+ * Builder class to create {@link PulsarTopic} instances.
+ *
+ * @author Chris Bono
+ * @author Alexander Preuß
+ */
+public class PulsarTopicBuilder {
+
+	private final String topicName;
+
+	private int numberOfPartitions;
+
+	protected PulsarTopicBuilder(String topicName) {
+		this.topicName = topicName;
+	}
+
+	/**
+	 * Sets the number of topic partitions.
+	 * @param numberOfPartitions the number of topic partitions
+	 * @return this builder
+	 */
+	public PulsarTopicBuilder numberOfPartitions(int numberOfPartitions) {
+		this.numberOfPartitions = numberOfPartitions;
+		return this;
+	}
+
+	/**
+	 * Constructs the {@link PulsarTopic} with the properties configured in this builder.
+	 * @return {@link PulsarTopic}
+	 */
+	public PulsarTopic build() {
+		return new PulsarTopic(this.topicName, this.numberOfPartitions);
+	}
+
+}

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

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2022 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+/**
+ * Tests for {@link PulsarAdministration}.
+ *
+ * @author Chris Bono
+ * @author Alexander Preuß
+ */
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration
+public class PulsarAdministrationTests extends AbstractContainerBaseTests {
+
+	private static final String NAMESPACE = "public/default";
+
+	@Autowired
+	private PulsarAdmin pulsarAdminClient;
+
+	@Autowired
+	private PulsarAdministration pulsarAdministration;
+
+	private void assertThatTopicsExist(List<PulsarTopic> expected) throws PulsarAdminException {
+		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);
+				}
+			}
+			else {
+				consumer.accept(topic.toString());
+			}
+
+		}).toList();
+		assertThat(pulsarAdminClient.topics().getList(NAMESPACE)).containsAll(expectedTopics);
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	static class AdminConfiguration {
+
+		@Bean
+		PulsarAdmin pulsarAdminClient() throws PulsarClientException {
+			return PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build();
+		}
+
+		@Bean
+		PulsarAdministration pulsarAdministration() {
+			return new PulsarAdministration(PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()));
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = CreateMissingTopicsTest.CreateMissingTopicsConfig.class)
+	class CreateMissingTopicsTest {
+
+		@Test
+		void topicsExist(@Autowired ObjectProvider<PulsarTopic> expectedTopics) throws Exception {
+			assertThatTopicsExist(expectedTopics.stream().toList());
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class CreateMissingTopicsConfig {
+
+			@Bean
+			PulsarTopic nonPartitionedTopic() {
+				return PulsarTopic.builder("cmt-non-partitioned-1").build();
+			}
+
+			@Bean
+			PulsarTopic nonPartitionedTopic2() {
+				return PulsarTopic.builder("cmt-non-partitioned-2").build();
+			}
+
+			@Bean
+			PulsarTopic partitionedTopic() {
+				return PulsarTopic.builder("cmt-partitioned-1").numberOfPartitions(4).build();
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = IncrementPartitionCountTest.IncrementPartitionCountConfig.class)
+	class IncrementPartitionCountTest {
+
+		@Test
+		void topicsExist(@Autowired ObjectProvider<PulsarTopic> expectedTopics) throws Exception {
+			assertThatTopicsExist(expectedTopics.stream().toList());
+			PulsarTopic biggerTopic = PulsarTopic.builder("ipc-partitioned-1").numberOfPartitions(4).build();
+			pulsarAdministration.createOrModifyTopics(biggerTopic);
+			assertThatTopicsExist(Collections.singletonList(biggerTopic));
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class IncrementPartitionCountConfig {
+
+			@Bean
+			PulsarTopic smallerTopic() {
+				return PulsarTopic.builder("ipc-partitioned-1").numberOfPartitions(1).build();
+			}
+
+		}
+
+	}
+
+	@Nested
+	@ContextConfiguration(classes = DecrementPartitionCountTest.DecrementPartitionCountConfig.class)
+	class DecrementPartitionCountTest {
+
+		@Test
+		void topicModificationThrows(@Autowired ObjectProvider<PulsarTopic> expectedTopics) throws Exception {
+			assertThatTopicsExist(expectedTopics.stream().toList());
+			PulsarTopic smallerTopic = PulsarTopic.builder("dpc-partitioned-1").numberOfPartitions(4).build();
+			assertThatIllegalStateException().isThrownBy(() -> pulsarAdministration.createOrModifyTopics(smallerTopic))
+					.withMessage(
+							"Topic persistent://public/default/dpc-partitioned-1 found with 8 partitions. Needs to be deleted first.");
+
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		static class DecrementPartitionCountConfig {
+
+			@Bean
+			PulsarTopic biggerTopic() {
+				return PulsarTopic.builder("dpc-partitioned-1").numberOfPartitions(8).build();
+			}
+
+		}
+
+	}
+
+}

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2022 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 static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.stream.Stream;
+
+import org.apache.pulsar.common.naming.TopicDomain;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Tests for {@link PulsarTopic}.
+ *
+ * @author Chris Bono
+ * @author Alexander Preuß
+ */
+public class PulsarTopicTests {
+
+	@Test
+	void builderDefaultValues() {
+		String topicName = "test-default-values";
+		PulsarTopicBuilder builder = PulsarTopic.builder(topicName);
+		PulsarTopic topic = builder.build();
+
+		assertThat(topic.topicName()).isEqualTo(topicName);
+		assertThat(topic.numberOfPartitions()).isEqualTo(0);
+	}
+
+	@ParameterizedTest
+	@MethodSource("topicComponentsProvider")
+	void topicComponents(PulsarTopic topic, TopicDomain domain, String tenant, String namespace, String topicName) {
+		PulsarTopic.TopicComponents components = topic.getComponents();
+		assertThat(components.domain()).isEqualTo(domain);
+		assertThat(components.tenant()).isEqualTo(tenant);
+		assertThat(components.namespace()).isEqualTo(namespace);
+		assertThat(components.name()).isEqualTo(topicName);
+	}
+
+	private static Stream<Arguments> topicComponentsProvider() {
+		return Stream.of(
+				Arguments.of(PulsarTopic.builder("topic-1").build(), TopicDomain.persistent, "public", "default",
+						"topic-1"),
+				Arguments.of(PulsarTopic.builder("public/default/topic-2").build(), TopicDomain.persistent, "public",
+						"default", "topic-2"),
+				Arguments.of(PulsarTopic.builder("persistent://public/default/topic-3").build(), TopicDomain.persistent,
+						"public", "default", "topic-3"),
+				Arguments.of(PulsarTopic.builder("public/my-namespace/topic-4").build(), TopicDomain.persistent,
+						"public", "my-namespace", "topic-4"),
+				Arguments.of(PulsarTopic.builder("my-tenant/my-namespace/topic-5").build(), TopicDomain.persistent,
+						"my-tenant", "my-namespace", "topic-5"),
+				Arguments.of(PulsarTopic.builder("non-persistent://public/my-namespace/topic-6").build(),
+						TopicDomain.non_persistent, "public", "my-namespace", "topic-6"),
+				Arguments.of(PulsarTopic.builder("non-persistent://my-tenant/my-namespace/topic-7").build(),
+						TopicDomain.non_persistent, "my-tenant", "my-namespace", "topic-7"));
+	}
+
+}