diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java new file mode 100644 index 00000000..ce47d0a4 --- /dev/null +++ b/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 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 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> getTopicsPerNamespace(Collection topics) { + return topics.stream().collect(Collectors.groupingBy(this::getTopicNamespaceIdentifier)); + } + + private String getTopicNamespaceIdentifier(PulsarTopic topic) { + return topic.getComponents().tenant() + "/" + topic.getComponents().namespace(); + } + + private List getMatchingTopicPartitions(PulsarTopic topic, List existingTopics) { + return existingTopics.stream().filter(existing -> existing.startsWith(topic + "-partition-")).toList(); + } + + private void createOrModifyTopicsIfNeeded(Collection 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 topics) { + Map> topicsPerNamespace = getTopicsPerNamespace(topics); + + Set topicsToCreate = new HashSet<>(); + Set topicsToModify = new HashSet<>(); + + topicsPerNamespace.forEach((namespace, requestedTopics) -> { + try { + List existingTopicsInNamespace = admin.topics().getList(namespace); + + for (PulsarTopic topic : requestedTopics) { + if (topic.isPartitioned()) { + List 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 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 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()); + } + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministrationOperations.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministrationOperations.java new file mode 100644 index 00000000..4cda0739 --- /dev/null +++ b/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); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java new file mode 100644 index 00000000..83480e24 --- /dev/null +++ b/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: + * + * 
{@code
+ * 	PulsarTopic topic = PulsarTopic.builder("topic-name").build();
+ * }
+ * @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) { + + } +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopicBuilder.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopicBuilder.java new file mode 100644 index 00000000..aab1bd9c --- /dev/null +++ b/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); + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java new file mode 100644 index 00000000..1aafc69f --- /dev/null +++ b/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 expected) throws PulsarAdminException { + List expectedTopics = expected.stream().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 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 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 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(); + } + + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicTests.java new file mode 100644 index 00000000..1dca55ce --- /dev/null +++ b/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 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")); + } + +}