From 6d23378fbb57a50c3a302be136f9173d2854085d Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 12 Aug 2024 14:03:32 -0500 Subject: [PATCH] Add support for default tenant and namespace (#766) See #756 --- .../inttest/app/ImperativeAppConfig.java | 3 +- .../pulsar/inttest/app/ReactiveAppConfig.java | 3 +- .../DefaultTenantAndNamespaceTests.java | 115 ++++++++++++++++++ .../inttest/config/ImperativeAppConfig.java | 110 +++++++++++++++++ .../inttest/config/ReactiveAppConfig.java | 115 ++++++++++++++++++ .../src/intTest/resources/logback-test.xml | 1 + .../reference/default-tenant-namespace.adoc | 28 +++++ .../ROOT/pages/reference/pulsar-admin.adoc | 13 +- .../antora/modules/ROOT/pages/whats-new.adoc | 17 +++ .../DefaultReactivePulsarConsumerFactory.java | 32 ++++- .../DefaultReactivePulsarReaderFactory.java | 32 ++++- .../DefaultReactivePulsarSenderFactory.java | 35 +++++- ...ultReactivePulsarConsumerFactoryTests.java | 25 ++++ ...faultReactivePulsarReaderFactoryTests.java | 20 +++ ...faultReactivePulsarSenderFactoryTests.java | 15 +++ .../ReactivePulsarListenerTestsBase.java | 5 +- .../build.gradle | 3 +- .../java/com/example/FailoverConsumerApp.java | 3 +- .../build.gradle | 2 +- .../ImperativeProduceAndConsumeApp.java | 3 +- .../core/DefaultPulsarConsumerFactory.java | 18 +++ .../core/DefaultPulsarProducerFactory.java | 19 ++- .../core/DefaultPulsarReaderFactory.java | 18 +++ .../pulsar/core/PulsarAdministration.java | 18 ++- .../pulsar/core/PulsarTopic.java | 81 ++++++++---- .../pulsar/core/PulsarTopicBuilder.java | 97 ++++++++++++++- .../CachingPulsarProducerFactoryTests.java | 28 ++++- .../DefaultPulsarConsumerFactoryTests.java | 27 ++++ .../DefaultPulsarProducerFactoryTests.java | 8 +- .../core/DefaultPulsarReaderFactoryTests.java | 28 +++++ .../PulsarAdministrationIntegrationTests.java | 34 +++--- .../core/PulsarProducerFactoryTests.java | 29 ++++- .../pulsar/core/PulsarTopicBuilderTests.java | 83 +++++++++++++ .../pulsar/core/PulsarTopicTests.java | 104 ++++++++++------ .../listener/PulsarListenerTestsBase.java | 5 +- src/checkstyle/checkstyle-suppressions.xml | 1 + 36 files changed, 1049 insertions(+), 129 deletions(-) create mode 100644 integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/DefaultTenantAndNamespaceTests.java create mode 100644 integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ImperativeAppConfig.java create mode 100644 integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ReactiveAppConfig.java create mode 100644 spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/default-tenant-namespace.adoc create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicBuilderTests.java diff --git a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ImperativeAppConfig.java b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ImperativeAppConfig.java index 830a9237..5e6fbe82 100644 --- a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ImperativeAppConfig.java +++ b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ImperativeAppConfig.java @@ -27,6 +27,7 @@ import org.springframework.context.annotation.Profile; import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.PulsarTopicBuilder; @SpringBootConfiguration @EnableAutoConfiguration @@ -39,7 +40,7 @@ class ImperativeAppConfig { @Bean PulsarTopic pulsarTestTopic() { - return PulsarTopic.builder(TOPIC).numberOfPartitions(1).build(); + return new PulsarTopicBuilder().name(TOPIC).numberOfPartitions(1).build(); } @Bean diff --git a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ReactiveAppConfig.java b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ReactiveAppConfig.java index c4857192..732027dd 100644 --- a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ReactiveAppConfig.java +++ b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ReactiveAppConfig.java @@ -26,6 +26,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Profile; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; @@ -43,7 +44,7 @@ class ReactiveAppConfig { @Bean PulsarTopic pulsarTestTopic() { - return PulsarTopic.builder(TOPIC).numberOfPartitions(1).build(); + return new PulsarTopicBuilder().name(TOPIC).numberOfPartitions(1).build(); } @Bean diff --git a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/DefaultTenantAndNamespaceTests.java b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/DefaultTenantAndNamespaceTests.java new file mode 100644 index 00000000..791a99ac --- /dev/null +++ b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/DefaultTenantAndNamespaceTests.java @@ -0,0 +1,115 @@ +/* + * Copyright 2012-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.inttest.config; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.function.Function; +import java.util.stream.IntStream; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.PulsarContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.context.ActiveProfiles; + +@Testcontainers(disabledWithoutDocker = true) +@ExtendWith(OutputCaptureExtension.class) +class DefaultTenantAndNamespaceTests { + + @SuppressWarnings("unused") + @Container + @ServiceConnection + static PulsarContainer PULSAR_CONTAINER = new PulsarContainer(PulsarTestContainerSupport.getPulsarImage()); + + @Nested + @SpringBootTest(classes = ImperativeAppConfig.class) + @ExtendWith(OutputCaptureExtension.class) + @ActiveProfiles("inttest.pulsar.imperative") + class WithImperativeApp { + + @Test + void produceConsumeWithDefaultTenantNamespace(CapturedOutput output, + @Autowired PulsarAdministration pulsarAdmin) { + TestVerifyUtils.verifyProduceConsume(output, 10, (i) -> ImperativeAppConfig.MSG_PREFIX + i); + TestVerifyUtils.verifyTopicsLocatedInTenantAndNamespace(pulsarAdmin, ImperativeAppConfig.TENANT, + ImperativeAppConfig.NAMESPACE, ImperativeAppConfig.NFQ_TOPIC); + } + + } + + @Nested + @SpringBootTest(classes = ReactiveAppConfig.class) + @ExtendWith(OutputCaptureExtension.class) + @ActiveProfiles("inttest.pulsar.reactive") + class WithReactiveApp { + + @Test + void produceConsumeWithDefaultTenantNamespace(CapturedOutput output, + @Autowired PulsarAdministration pulsarAdmin) { + TestVerifyUtils.verifyProduceConsume(output, 10, (i) -> ReactiveAppConfig.MSG_PREFIX + i); + TestVerifyUtils.verifyTopicsLocatedInTenantAndNamespace(pulsarAdmin, ReactiveAppConfig.TENANT, + ReactiveAppConfig.NAMESPACE, ReactiveAppConfig.NFQ_TOPIC); + } + + } + + private static class TestVerifyUtils { + + static void verifyProduceConsume(CapturedOutput output, int numExpectedMessages, + Function expectedMessageFactory) { + var expectedOutput = new ArrayList(); + IntStream.range(0, numExpectedMessages).forEachOrdered((i) -> { + var expectedMsg = expectedMessageFactory.apply(i); + expectedOutput.add("++++++PRODUCE %s------".formatted(expectedMsg)); + expectedOutput.add("++++++CONSUME %s------".formatted(expectedMsg)); + }); + Awaitility.waitAtMost(Duration.ofSeconds(15)) + .untilAsserted(() -> assertThat(output).contains(expectedOutput)); + } + + static void verifyTopicsLocatedInTenantAndNamespace(PulsarAdministration pulsarAdmin, String tenant, + String namespace, String topic) { + // verify topics created in expected tenant/namespace and not in + // public/default + try (var admin = pulsarAdmin.createAdminClient()) { + var fqTopic = "persistent://%s/%s/%s".formatted(tenant, namespace, topic); + assertThat(admin.topics().getList("%s/%s".formatted(tenant, namespace))).containsExactly(fqTopic); + assertThat(admin.topics().getList("public/default")) + .noneSatisfy(t -> assertThat(t).doesNotEndWith("/" + topic)); + } + catch (Exception ex) { + throw new RuntimeException(ex); + } + } + + } + +} diff --git a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ImperativeAppConfig.java b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ImperativeAppConfig.java new file mode 100644 index 00000000..aaba9ec2 --- /dev/null +++ b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ImperativeAppConfig.java @@ -0,0 +1,110 @@ +/* + * 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.inttest.config; + +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.common.naming.TopicDomain; +import org.apache.pulsar.common.policies.data.TenantInfoImpl; + +import org.springframework.boot.ApplicationRunner; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Profile; +import org.springframework.pulsar.annotation.PulsarListener; +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; +import org.springframework.pulsar.core.PulsarTopicBuilder; +import org.springframework.pulsar.core.TopicResolver; + +@SpringBootConfiguration +@EnableAutoConfiguration +@Profile("inttest.pulsar.imperative") +class ImperativeAppConfig { + + private static final Log LOG = LogFactory.getLog(ImperativeAppConfig.class); + static final String TENANT = "my-tenant-i"; + static final String NAMESPACE = "my-namespace-i"; + static final String NFQ_TOPIC = "dtant-topic-i"; + static final String FQ_TOPIC = "persistent://my-tenant-i/my-namespace-i/dtant-topic-i"; + static final String MSG_PREFIX = "DefaultTenantNamespace-i:"; + + @Bean + PulsarTopicBuilder topicBuilder() { + return new PulsarTopicBuilder(TopicDomain.persistent, TENANT, NAMESPACE); + } + + @Bean + PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver, + PulsarTopicBuilder topicBuilder) { + var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, null, null, topicResolver); + producerFactory.setTopicBuilder(topicBuilder); + return producerFactory; + } + + @Bean + PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, PulsarTopicBuilder topicBuilder) { + var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null); + consumerFactory.setTopicBuilder(topicBuilder); + return consumerFactory; + } + + @PulsarListener(topics = NFQ_TOPIC) + void consumeFromNonFullyQualifiedTopic(String msg) { + LOG.info("++++++CONSUME %s------".formatted(msg)); + } + + @PulsarListener(topics = FQ_TOPIC) + void consumeFromFullyQualifiedTopic(String msg) { + LOG.info("++++++CONSUME %s------".formatted(msg)); + } + + @Bean + ApplicationRunner produceWithDefaultTenantAndNamespace(PulsarAdministration pulsarAdmin, + PulsarTemplate template) { + createTenantAndNamespace(pulsarAdmin); + return (args) -> { + for (int i = 0; i < 10; i++) { + var msg = MSG_PREFIX + i; + template.send((i < 5) ? FQ_TOPIC : NFQ_TOPIC, msg); + LOG.info("++++++PRODUCE %s------".formatted(msg)); + } + }; + } + + private void createTenantAndNamespace(PulsarAdministration pulsarAdmin) { + try (var admin = pulsarAdmin.createAdminClient()) { + admin.tenants() + .createTenant(TENANT, TenantInfoImpl.builder().allowedClusters(Set.of("standalone")).build()); + LOG.info("Created tenant -> %s".formatted(admin.tenants().getTenantInfo(TENANT))); + admin.namespaces().createNamespace("%s/%s".formatted(TENANT, NAMESPACE)); + LOG.info("Created namespace -> %s".formatted(admin.namespaces().getNamespaces(TENANT))); + } + catch (Exception ex) { + throw new RuntimeException(ex); + } + } + +} diff --git a/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ReactiveAppConfig.java b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ReactiveAppConfig.java new file mode 100644 index 00000000..afb45e43 --- /dev/null +++ b/integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ReactiveAppConfig.java @@ -0,0 +1,115 @@ +/* + * 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.inttest.config; + +import java.util.List; +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.pulsar.common.naming.TopicDomain; +import org.apache.pulsar.common.policies.data.TenantInfoImpl; +import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; + +import org.springframework.boot.ApplicationRunner; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Profile; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.core.PulsarTopicBuilder; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; + +import reactor.core.publisher.Mono; + +@SpringBootConfiguration +@EnableAutoConfiguration +@Profile("inttest.pulsar.reactive") +class ReactiveAppConfig { + + private static final Log LOG = LogFactory.getLog(ReactiveAppConfig.class); + static final String TENANT = "my-tenant-r"; + static final String NAMESPACE = "my-namespace-r"; + static final String NFQ_TOPIC = "dtant-topic-r"; + static final String FQ_TOPIC = "persistent://my-tenant-r/my-namespace-r/dtant-topic-r"; + static final String MSG_PREFIX = "DefaultTenantNamespace-r:"; + + @Bean + PulsarTopicBuilder topicBuilder() { + return new PulsarTopicBuilder(TopicDomain.persistent, TENANT, NAMESPACE); + } + + @Bean + ReactivePulsarSenderFactory reactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient, + PulsarTopicBuilder topicBuilder) { + return DefaultReactivePulsarSenderFactory.builderFor(reactivePulsarClient) + .withTopicBuilder(topicBuilder) + .build(); + } + + @Bean + ReactivePulsarConsumerFactory reactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient, + PulsarTopicBuilder topicBuilder) { + var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, List.of()); + consumerFactory.setTopicBuilder(topicBuilder); + return consumerFactory; + } + + @ReactivePulsarListener(topics = NFQ_TOPIC) + Mono consumeFromNonFullyQualifiedTopic(String msg) { + LOG.info("++++++CONSUME %s------".formatted(msg)); + return Mono.empty(); + } + + @ReactivePulsarListener(topics = FQ_TOPIC) + Mono consumeFromFullyQualifiedTopic(String msg) { + LOG.info("++++++CONSUME %s------".formatted(msg)); + return Mono.empty(); + } + + @Bean + ApplicationRunner produceWithDefaultTenantAndNamespace(PulsarAdministration pulsarAdmin, + ReactivePulsarTemplate template) { + createTenantAndNamespace(pulsarAdmin); + return (args) -> { + for (int i = 0; i < 10; i++) { + var msg = MSG_PREFIX + i; + template.send((i < 5) ? FQ_TOPIC : NFQ_TOPIC, msg).subscribe(); + LOG.info("++++++PRODUCE %s------".formatted(msg)); + } + }; + } + + private void createTenantAndNamespace(PulsarAdministration pulsarAdmin) { + try (var admin = pulsarAdmin.createAdminClient()) { + admin.tenants() + .createTenant(TENANT, TenantInfoImpl.builder().allowedClusters(Set.of("standalone")).build()); + LOG.info("Created tenant -> %s".formatted(admin.tenants().getTenantInfo(TENANT))); + admin.namespaces().createNamespace("%s/%s".formatted(TENANT, NAMESPACE)); + LOG.info("Created namespace -> %s".formatted(admin.namespaces().getNamespaces(TENANT))); + } + catch (Exception ex) { + throw new RuntimeException(ex); + } + } + +} diff --git a/integration-tests/src/intTest/resources/logback-test.xml b/integration-tests/src/intTest/resources/logback-test.xml index a2c4c500..cd25bb7d 100644 --- a/integration-tests/src/intTest/resources/logback-test.xml +++ b/integration-tests/src/intTest/resources/logback-test.xml @@ -11,4 +11,5 @@ + diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/default-tenant-namespace.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/default-tenant-namespace.adoc new file mode 100644 index 00000000..b810d8a3 --- /dev/null +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/default-tenant-namespace.adoc @@ -0,0 +1,28 @@ +[[default-tenant-namespace]] += Default Tenant / Namespace +include::../attributes/attributes.adoc[] + +Pulsar has built-in support for {apache-pulsar-docs}/concepts-multi-tenancy/[multi-tenancy]. +When producing or consuming messages in Pulsar, the specified topic is actually a topic URL of the following format: +``` +(persistent|non-persistent)://tenant/namespace/topic +``` +The URL dictates which tenant and namespace the operation is targeted against. +However, when not fully-qualified (i.e. only topic name is specified), the default tenant of `public` and namespace of `default` is used. + +Spring for Apache Pulsar allows you to specify a default tenant and/or namespace to use when producing or consuming messages against a non-fully-qualified topic URL. + +[discrete] +== Configuration + +[discrete] +=== With Spring Boot +When using the Spring Boot you can simply set the {spring-boot-pulsar-config-props}[`spring.pulsar.defaults.tenant`] and {spring-boot-pulsar-config-props}[`spring.pulsar.defaults.namespace`] application properties to specify these defaults. + +[discrete] +=== Without Spring Boot +However, if you are instead manually configuring the components, you can provide a `PulsarTopicBuilder` configured with the desired default topic and namespace when constructing the corresponding producer or consumer factory. +All default consumer/reader/producer factory implementations (imperative and reactive) allow a topic builder to be specified. + +[NOTE] +You will need to specify the topic builder on each manually configured factory that you want to use the default tenant/namespace diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc index 3f8f5f06..6a6ed031 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc @@ -28,17 +28,20 @@ For all such beans, the `PulsarAdministration` either creates the corresponding The following example shows how to add `PulsarTopic` beans to let the `PulsarAdministration` auto-create topics for you: -[source,java] +[source,java,indent=0,subs="verbatim"] ---- @Bean PulsarTopic simpleTopic { - // This will create a non-partitioned topic in the public/default namespace - return PulsarTopic.builder("simple-topic").build(); + // This will create a non-partitioned persistent topic in the 'public/default' tenant/namespace + return new PulsarTopicBuilder().name("my-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(); + // This will create a persistent topic with 3 partitions in the provided tenant and namespace + return new PulsarTopicBuilder() + .name("persistent://my-tenant/my-namespace/partitioned-topic") + .numberOfPartitions(3) + .build(); } ---- diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc index 2c054a8d..d5ae81c3 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc @@ -10,11 +10,28 @@ This section covers the changes made from version 1.1 to version 1.2. You can provide your own Jackson `ObjectMapper` that Pulsar will use when producing and consuming JSON messages. See xref:./reference/custom-object-mapper.adoc[Custom Object Mapper] for more details. +=== Default Tenant and Namespace +You can specify a default tenant and/or namespace to use when producing or consuming messages against a non-fully-qualified topic URL. +See xref:./reference/default-tenant-namespace.adoc[Default Tenant / Namespace] for more details. + === Deprecations ==== PulsarClient#getPartitionsForTopic(java.lang.String) Version `3.3.1` of the Pulsar client deprecates the `getPartitionsForTopic(java.lang.String)` in favor of `getPartitionsForTopic(java.lang.String, boolean metadataAutoCreationEnabled)`. +==== PulsarTopic#builder +The `PulsarTopicBuilder` is now a registered bean that is configured with default values for domain, tenant, and namespace. +As such, this convenience method is no longer needed. +Instead, inject the builder bean where needed. + +=== Breaking Changes + +==== PulsarTopic# +The `PulsarTopic` constructor now requires a fully qualified topic name (`domain://tenant/namespace/name`). +If you are invoking the constructor you will need to be sure the topic you pass in is fully-qualified. +A better alternative is to instead use the `PulsarTopicBuilder` as it does not require fully qualified names and will add default values for the missing components in the specified name. + + [[what-s-new-in-1-1-since-1-0]] == What's New in 1.1 Since 1.0 :page-section-summary-toc: 1 diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java index 8a51a675..e03489bd 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java @@ -25,6 +25,7 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder; import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; import org.springframework.lang.Nullable; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.util.CollectionUtils; /** @@ -41,6 +42,9 @@ public class DefaultReactivePulsarConsumerFactory implements ReactivePulsarCo @Nullable private final List> defaultConfigCustomizers; + @Nullable + private PulsarTopicBuilder topicBuilder; + /** * Construct an instance. * @param reactivePulsarClient the reactive client @@ -53,6 +57,18 @@ public class DefaultReactivePulsarConsumerFactory implements ReactivePulsarCo this.defaultConfigCustomizers = defaultConfigCustomizers; } + /** + * Non-fully-qualified topic names specified on the created consumers will be + * automatically fully-qualified with a default prefix + * ({@code domain://tenant/namespace}) according to the specified topic builder. + * @param topicBuilder the topic builder used to fully qualify topic names or null to + * not fully qualify topic names + * @since 1.2.0 + */ + public void setTopicBuilder(@Nullable PulsarTopicBuilder topicBuilder) { + this.topicBuilder = topicBuilder; + } + @Override public ReactiveMessageConsumer createConsumer(Schema schema) { return createConsumer(schema, Collections.emptyList()); @@ -61,20 +77,28 @@ public class DefaultReactivePulsarConsumerFactory implements ReactivePulsarCo @Override public ReactiveMessageConsumer createConsumer(Schema schema, List> customizers) { - ReactiveMessageConsumerBuilder consumerBuilder = this.reactivePulsarClient.messageConsumer(schema); - // Apply the default customizers if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) { this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(consumerBuilder))); } - // Apply the user specified customizers if (!CollectionUtils.isEmpty(customizers)) { customizers.forEach((c) -> c.customize(consumerBuilder)); } - + if (this.topicBuilder != null) { + this.ensureTopicNamesFullyQualified(consumerBuilder); + } return consumerBuilder.build(); } + protected void ensureTopicNamesFullyQualified(ReactiveMessageConsumerBuilder consumerBuilder) { + var mutableSpec = consumerBuilder.getMutableSpec(); + var topics = mutableSpec.getTopicNames(); + if (!CollectionUtils.isEmpty(topics)) { + var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList(); + mutableSpec.setTopicNames(fullyQualifiedTopics); + } + } + } diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java index 374f39ce..5af71131 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java @@ -25,6 +25,7 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder; import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; import org.springframework.lang.Nullable; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.util.CollectionUtils; /** @@ -41,6 +42,9 @@ public class DefaultReactivePulsarReaderFactory implements ReactivePulsarRead @Nullable private final List> defaultConfigCustomizers; + @Nullable + private PulsarTopicBuilder topicBuilder; + /** * Construct an instance. * @param reactivePulsarClient the reactive client @@ -53,6 +57,18 @@ public class DefaultReactivePulsarReaderFactory implements ReactivePulsarRead this.defaultConfigCustomizers = defaultConfigCustomizers; } + /** + * Non-fully-qualified topic names specified on the created readers will be + * automatically fully-qualified with a default prefix + * ({@code domain://tenant/namespace}) according to the specified topic builder. + * @param topicBuilder the topic builder used to fully qualify topic names or null to + * not fully qualify topic names + * @since 1.2.0 + */ + public void setTopicBuilder(@Nullable PulsarTopicBuilder topicBuilder) { + this.topicBuilder = topicBuilder; + } + @Override public ReactiveMessageReader createReader(Schema schema) { return createReader(schema, Collections.emptyList()); @@ -61,20 +77,28 @@ public class DefaultReactivePulsarReaderFactory implements ReactivePulsarRead @Override public ReactiveMessageReader createReader(Schema schema, List> customizers) { - ReactiveMessageReaderBuilder readerBuilder = this.reactivePulsarClient.messageReader(schema); - // Apply the default customizers if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) { this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(readerBuilder))); } - // Apply the user specified customizers if (!CollectionUtils.isEmpty(customizers)) { customizers.forEach((c) -> c.customize(readerBuilder)); } - + if (this.topicBuilder != null) { + this.ensureTopicNamesFullyQualified(readerBuilder); + } return readerBuilder.build(); } + protected void ensureTopicNamesFullyQualified(ReactiveMessageReaderBuilder readerBuilder) { + var mutableSpec = readerBuilder.getMutableSpec(); + var topics = mutableSpec.getTopicNames(); + if (!CollectionUtils.isEmpty(topics)) { + var fullyQualifiedTopics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList(); + mutableSpec.setTopicNames(fullyQualifiedTopics); + } + } + } diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java index a7f660c4..b76dcdc0 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java @@ -32,6 +32,7 @@ import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; import org.springframework.pulsar.core.DefaultTopicResolver; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.pulsar.core.TopicResolver; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -65,14 +66,19 @@ public final class DefaultReactivePulsarSenderFactory @Nullable private final List> defaultConfigCustomizers; + @Nullable + private final PulsarTopicBuilder topicBuilder; + private DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient, TopicResolver topicResolver, @Nullable ReactiveMessageSenderCache reactiveMessageSenderCache, @Nullable String defaultTopic, - @Nullable List> defaultConfigCustomizers) { + @Nullable List> defaultConfigCustomizers, + @Nullable PulsarTopicBuilder topicBuilder) { this.reactivePulsarClient = reactivePulsarClient; this.topicResolver = topicResolver; this.reactiveMessageSenderCache = reactiveMessageSenderCache; this.defaultTopic = defaultTopic; this.defaultConfigCustomizers = defaultConfigCustomizers; + this.topicBuilder = topicBuilder; } /** @@ -116,7 +122,7 @@ public final class DefaultReactivePulsarSenderFactory private ReactiveMessageSender doCreateReactiveMessageSender(Schema schema, @Nullable String topic, @Nullable List> customizers) { Objects.requireNonNull(schema, "Schema must be specified"); - String resolvedTopic = this.topicResolver.resolveTopic(topic, () -> getDefaultTopic()).orElseThrow(); + String resolvedTopic = this.resolveTopicName(topic); this.logger.trace(() -> "Creating reactive message sender for '%s' topic".formatted(resolvedTopic)); ReactiveMessageSenderBuilder sender = this.reactivePulsarClient.messageSender(schema); @@ -140,6 +146,12 @@ public final class DefaultReactivePulsarSenderFactory return sender.build(); } + protected String resolveTopicName(String userSpecifiedTopic) { + var resolvedTopic = this.topicResolver.resolveTopic(userSpecifiedTopic, this::getDefaultTopic).orElseThrow(); + return this.topicBuilder != null ? this.topicBuilder.getFullyQualifiedNameForTopic(resolvedTopic) + : resolvedTopic; + } + @Override public String getDefaultTopic() { return this.defaultTopic; @@ -192,6 +204,9 @@ public final class DefaultReactivePulsarSenderFactory private TopicResolver topicResolver = new DefaultTopicResolver(); + @Nullable + private PulsarTopicBuilder topicBuilder; + @Nullable private ReactiveMessageSenderCache messageSenderCache; @@ -216,6 +231,20 @@ public final class DefaultReactivePulsarSenderFactory return this; } + /** + * Provide the topic builder to use to fully qualify topic names. + * Non-fully-qualified topic names specified on the created senders will be + * automatically fully-qualified with a default prefix + * ({@code domain://tenant/namespace}) according to the topic builder. + * @param topicBuilder the topic builder to use + * @return this same builder instance + * @since 1.2.0 + */ + public Builder withTopicBuilder(PulsarTopicBuilder topicBuilder) { + this.topicBuilder = topicBuilder; + return this; + } + /** * Provide the message sender cache to use. * @param messageSenderCache the message sender cache to use @@ -266,7 +295,7 @@ public final class DefaultReactivePulsarSenderFactory public DefaultReactivePulsarSenderFactory build() { Assert.notNull(this.topicResolver, "Topic resolver is required"); return new DefaultReactivePulsarSenderFactory<>(this.reactivePulsarClient, this.topicResolver, - this.messageSenderCache, this.defaultTopic, this.defaultConfigCustomizers); + this.messageSenderCache, this.defaultTopic, this.defaultConfigCustomizers, this.topicBuilder); } } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java index 040d62a3..ea2c14d4 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.util.Collections; import java.util.List; @@ -31,6 +33,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.springframework.pulsar.core.PulsarTopicBuilder; + /** * Tests for {@link DefaultReactivePulsarConsumerFactory}. * @@ -104,4 +108,25 @@ class DefaultReactivePulsarConsumerFactoryTests { } + @Nested + class FactoryCreatedWithTopicBuilder { + + @Test + void createConsumer() { + var topicBuilder = spy(new PulsarTopicBuilder()); + var consumerFactory = new DefaultReactivePulsarConsumerFactory( + AdaptedReactivePulsarClientFactory.create((PulsarClient) null), null); + consumerFactory.setTopicBuilder(topicBuilder); + var inputTopic = "my-topic"; + var fullyQualifiedTopic = "persistent://public/default/my-topic"; + var consumer = consumerFactory.createConsumer(SCHEMA, + Collections.singletonList(builder -> builder.topic(inputTopic))); + assertThat(consumer) + .extracting("consumerSpec", InstanceOfAssertFactories.type(ReactiveMessageConsumerSpec.class)) + .hasFieldOrPropertyWithValue("topicNames", List.of(fullyQualifiedTopic)); + verify(topicBuilder).getFullyQualifiedNameForTopic(inputTopic); + } + + } + } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java index 08f60894..b06fc439 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.util.Collections; import java.util.List; @@ -29,6 +31,8 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderSpec; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; +import org.springframework.pulsar.core.PulsarTopicBuilder; + /** * Tests for {@link DefaultReactivePulsarReaderFactory}. * @@ -66,4 +70,20 @@ class DefaultReactivePulsarReaderFactoryTests { .isEqualTo("new-test-reader"); } + @Test + void createReaderUsingTopicBuilder() { + var inputTopic = "my-topic"; + var fullyQualifiedTopic = "persistent://public/default/my-topic"; + var topicBuilder = spy(new PulsarTopicBuilder()); + var readerFactory = new DefaultReactivePulsarReaderFactory( + AdaptedReactivePulsarClientFactory.create((PulsarClient) null), null); + readerFactory.setTopicBuilder(topicBuilder); + var reader = readerFactory.createReader(schema, + Collections.singletonList(builder -> builder.topic(inputTopic))); + assertThat(reader).extracting("readerSpec", InstanceOfAssertFactories.type(ReactiveMessageReaderSpec.class)) + .extracting(ReactiveMessageReaderSpec::getTopicNames) + .isEqualTo(List.of(fullyQualifiedTopic)); + verify(topicBuilder).getFullyQualifiedNameForTopic(inputTopic); + } + } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java index 31d573c8..c8b63463 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java @@ -46,6 +46,7 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.mockito.InOrder; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.pulsar.core.TopicResolver; /** @@ -74,6 +75,20 @@ class DefaultReactivePulsarSenderFactoryTests { assertThat(senderFactory).hasFieldOrPropertyWithValue("topicResolver", customTopicResolver); } + @Test + void createSenderWithTopicBuilder() { + var inputTopic = "my-topic"; + var fullyQualifiedTopic = "persistent://public/default/my-topic"; + var topicBuilder = spy(new PulsarTopicBuilder()); + var senderFactory = DefaultReactivePulsarSenderFactory.builderFor(mock(PulsarClient.class)) + .withTopicBuilder(topicBuilder) + .build(); + assertThat(senderFactory).hasFieldOrPropertyWithValue("topicBuilder", topicBuilder); + var sender = senderFactory.createSender(schema, inputTopic); + assertThatSenderHasTopic(sender, fullyQualifiedTopic); + verify(topicBuilder).getFullyQualifiedNameForTopic(inputTopic); + } + private void assertThatSenderHasTopic(ReactiveMessageSender sender, String expectedTopic) { assertThatSenderSpecSatisfies(sender, (senderSpec) -> assertThat(senderSpec).extracting(ReactiveMessageSenderSpec::getTopicName) diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTestsBase.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTestsBase.java index 06560b61..b9712fa4 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTestsBase.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTestsBase.java @@ -32,6 +32,7 @@ import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar; @@ -107,7 +108,9 @@ abstract class ReactivePulsarListenerTestsBase implements PulsarTestContainerSup @Bean PulsarTopic partitionedTopic() { - return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build(); + return new PulsarTopicBuilder().name("persistent://public/default/concurrency-on-pl") + .numberOfPartitions(3) + .build(); } @Bean diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle b/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle index c3dedc25..f796f69d 100644 --- a/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle @@ -25,7 +25,8 @@ dependencies { implementation 'io.zipkin.reporter2:zipkin-reporter-brave' implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection' developmentOnly 'org.springframework.boot:spring-boot-docker-compose' - + // TODO remove when new PulsarTopicBuilder published + implementation project(':spring-pulsar') testImplementation project(':spring-pulsar-test') testRuntimeOnly 'ch.qos.logback:logback-classic' testImplementation "org.springframework.boot:spring-boot-starter-test" diff --git a/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java index 03d5551a..a54c53f2 100644 --- a/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java +++ b/spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java @@ -31,6 +31,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.PulsarTopicBuilder; @SpringBootApplication public class FailoverConsumerApp { @@ -45,7 +46,7 @@ public class FailoverConsumerApp { @Bean PulsarTopic failoverDemoTopic() { - return PulsarTopic.builder(TOPIC).numberOfPartitions(3).build(); + return new PulsarTopicBuilder().name(TOPIC).numberOfPartitions(3).build(); } @Bean diff --git a/spring-pulsar-sample-apps/sample-imperative-produce-consume/build.gradle b/spring-pulsar-sample-apps/sample-imperative-produce-consume/build.gradle index 83b8eac7..efecf96c 100644 --- a/spring-pulsar-sample-apps/sample-imperative-produce-consume/build.gradle +++ b/spring-pulsar-sample-apps/sample-imperative-produce-consume/build.gradle @@ -21,7 +21,7 @@ ext['pulsar.version'] = "${pulsarVersion}" dependencies { implementation 'org.springframework.boot:spring-boot-starter-pulsar' developmentOnly 'org.springframework.boot:spring-boot-docker-compose' - // temporary until JsonSchemaUtil published + // TODO remove when new PulsarTopicBuilder published implementation project(':spring-pulsar') implementation(testFixtures(project(":spring-pulsar"))) implementation project(':spring-pulsar-test') diff --git a/spring-pulsar-sample-apps/sample-imperative-produce-consume/src/main/java/com/example/ImperativeProduceAndConsumeApp.java b/spring-pulsar-sample-apps/sample-imperative-produce-consume/src/main/java/com/example/ImperativeProduceAndConsumeApp.java index 7d5f0d66..13231975 100644 --- a/spring-pulsar-sample-apps/sample-imperative-produce-consume/src/main/java/com/example/ImperativeProduceAndConsumeApp.java +++ b/spring-pulsar-sample-apps/sample-imperative-produce-consume/src/main/java/com/example/ImperativeProduceAndConsumeApp.java @@ -30,6 +30,7 @@ import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.test.model.UserRecord; import org.springframework.pulsar.test.model.json.UserRecordObjectMapper; @@ -98,7 +99,7 @@ public class ImperativeProduceAndConsumeApp { @Bean PulsarTopic partitionedTopic() { - return PulsarTopic.builder(TOPIC).numberOfPartitions(3).build(); + return new PulsarTopicBuilder().name(TOPIC).numberOfPartitions(3).build(); } @Bean diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java index ce389b5d..fb2827e4 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java @@ -52,6 +52,9 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory @Nullable private final List> defaultConfigCustomizers; + @Nullable + private PulsarTopicBuilder topicBuilder; + /** * Construct a consumer factory instance. * @param pulsarClient the client used to consume @@ -64,6 +67,18 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory this.defaultConfigCustomizers = defaultConfigCustomizers; } + /** + * Non-fully-qualified topic names specified on the created consumers will be + * automatically fully-qualified with a default prefix + * ({@code domain://tenant/namespace}) according to the specified topic builder. + * @param topicBuilder the topic builder used to fully qualify topic names or null to + * not fully qualify topic names + * @since 1.2.0 + */ + public void setTopicBuilder(@Nullable PulsarTopicBuilder topicBuilder) { + this.topicBuilder = topicBuilder; + } + @Override public Consumer createConsumer(Schema schema, @Nullable Collection topics, @Nullable String subscriptionName, ConsumerBuilderCustomizer customizer) { @@ -111,6 +126,9 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory } private void replaceTopicsOnBuilder(ConsumerBuilder builder, Collection topics) { + if (this.topicBuilder != null) { + topics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList(); + } var builderImpl = (ConsumerBuilderImpl) builder; builderImpl.getConf().setTopicNames(new HashSet<>(topics)); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index c3b775eb..04851331 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -57,6 +57,9 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory private final TopicResolver topicResolver; + @Nullable + private PulsarTopicBuilder topicBuilder; + /** * Construct a producer factory that uses a default topic resolver. * @param pulsarClient the client used to create the producers @@ -102,6 +105,18 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory this.topicResolver = Objects.requireNonNull(topicResolver, "topicResolver must not be null"); } + /** + * Non-fully-qualified topic names specified on the created producers will be + * automatically fully-qualified with a default prefix + * ({@code domain://tenant/namespace}) according to the specified topic builder. + * @param topicBuilder the topic builder used to fully qualify topic names or null to + * not fully qualify topic names + * @since 1.2.0 + */ + public void setTopicBuilder(@Nullable PulsarTopicBuilder topicBuilder) { + this.topicBuilder = topicBuilder; + } + @Override public Producer createProducer(Schema schema, @Nullable String topic) { return doCreateProducer(schema, topic, null, null); @@ -169,7 +184,9 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory } protected String resolveTopicName(String userSpecifiedTopic) { - return this.topicResolver.resolveTopic(userSpecifiedTopic, this::getDefaultTopic).orElseThrow(); + var resolvedTopic = this.topicResolver.resolveTopic(userSpecifiedTopic, this::getDefaultTopic).orElseThrow(); + return this.topicBuilder != null ? this.topicBuilder.getFullyQualifiedNameForTopic(resolvedTopic) + : resolvedTopic; } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java index cb4034bb..44a8764c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java @@ -45,6 +45,9 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { @Nullable private final List> defaultConfigCustomizers; + @Nullable + private PulsarTopicBuilder topicBuilder; + /** * Construct a reader factory instance with no default configuration. * @param pulsarClient the client used to consume @@ -65,6 +68,18 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { this.defaultConfigCustomizers = defaultConfigCustomizers; } + /** + * Non-fully-qualified topic names specified on the created readers will be + * automatically fully-qualified with a default prefix + * ({@code domain://tenant/namespace}) according to the specified topic builder. + * @param topicBuilder the topic builder used to fully qualify topic names or null to + * not fully qualify topic names + * @since 1.2.0 + */ + public void setTopicBuilder(@Nullable PulsarTopicBuilder topicBuilder) { + this.topicBuilder = topicBuilder; + } + @Override public Reader createReader(@Nullable List topics, @Nullable MessageId messageId, Schema schema, @Nullable List> customizers) throws PulsarClientException { @@ -92,6 +107,9 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { } private void replaceTopicsOnBuilder(ReaderBuilder builder, Collection topics) { + if (this.topicBuilder != null) { + topics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList(); + } var builderImpl = (ReaderBuilderImpl) builder; builderImpl.getConf().setTopicNames(new HashSet<>(topics)); } 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 index f6d77e1a..8766b6aa 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java @@ -130,12 +130,13 @@ public class PulsarAdministration } private String getTopicNamespaceIdentifier(PulsarTopic topic) { - return topic.getComponents().tenant() + "/" + topic.getComponents().namespace(); + var components = topic.getComponents(); + return components.tenant() + "/" + components.namespace(); } private List getMatchingTopicPartitions(PulsarTopic topic, List existingTopics) { return existingTopics.stream() - .filter(existing -> existing.startsWith(topic.getFullyQualifiedTopicName() + "-partition-")) + .filter(existing -> existing.startsWith(topic.topicName() + "-partition-")) .toList(); } @@ -143,7 +144,6 @@ public class PulsarAdministration if (CollectionUtils.isEmpty(topics)) { return; } - try (PulsarAdmin admin = createAdminClient()) { doCreateOrModifyTopicsIfNeeded(admin, topics); } @@ -163,7 +163,7 @@ public class PulsarAdministration var existingTopicsInNamespace = admin.topics().getList(namespace); for (var topic : requestedTopics) { - var topicName = topic.getFullyQualifiedTopicName(); + var topicName = topic.topicName(); if (topic.isPartitioned()) { if (existingTopicsInNamespace.contains(topicName)) { throw new IllegalStateException( @@ -214,9 +214,8 @@ public class PulsarAdministration } private void createTopics(PulsarAdmin admin, Set topicsToCreate) throws PulsarAdminException { - this.logger.debug(() -> "Creating topics: " + topicsToCreate.stream() - .map(PulsarTopic::getFullyQualifiedTopicName) - .collect(Collectors.joining(","))); + this.logger.debug(() -> "Creating topics: " + + topicsToCreate.stream().map(PulsarTopic::topicName).collect(Collectors.joining(","))); for (var topic : topicsToCreate) { if (topic.isPartitioned()) { admin.topics().createPartitionedTopic(topic.topicName(), topic.numberOfPartitions()); @@ -228,9 +227,8 @@ public class PulsarAdministration } private void modifyTopics(PulsarAdmin admin, Set topicsToModify) throws PulsarAdminException { - this.logger.debug(() -> "Modifying topics: " + topicsToModify.stream() - .map(PulsarTopic::getFullyQualifiedTopicName) - .collect(Collectors.joining(","))); + this.logger.debug(() -> "Modifying topics: " + + topicsToModify.stream().map(PulsarTopic::topicName).collect(Collectors.joining(","))); for (var topic : topicsToModify) { admin.topics().updatePartitionedTopic(topic.topicName(), topic.numberOfPartitions()); } 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 index 76479962..57096383 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopic.java @@ -16,25 +16,57 @@ package org.springframework.pulsar.core; +import java.util.regex.Pattern; + import org.apache.pulsar.common.naming.TopicDomain; +import org.springframework.util.Assert; + /** - * Model class for a Pulsar topic. + * Represents a Pulsar topic. + *

+ * The input {@code topicName} must be fully-qualified. As such, it is recommended to use + * the {@link PulsarTopicBuilder} to create instances like this: 

{@code
+ * 	PulsarTopic topic = new PulsarTopicBuilder().name("my-topic").build();
+ * }
The builder is more lenient and allows non-fully-qualified topic names to be + * input and fully qualifies the output name using its configured default tenant and + * namepsace. * - * Use the {@link PulsarTopicBuilder} to create instances like this: - * - * 
{@code
- * 	PulsarTopic topic = PulsarTopic.builder("topic-name").build();
- * }
- * - * @param topicName the topic name + * @param topicName the fully qualified topic name in the format + * {@code 'domain://tenant/namespace/name'} * @param numberOfPartitions the number of partitions, or 0 for non-partitioned topics * @author Alexander Preuß + * @author Chris Bono + * @see PulsarTopicBuilder */ public record PulsarTopic(String topicName, int numberOfPartitions) { + // Pulsar allows (a-zA-Z_0-9) and special chars -=:. for names + private static final String NAME_PATTERN_STR = "[-=:\\.\\w]*"; + + private static Pattern TOPIC_NAME_PATTERN = Pattern.compile("(persistent|non-persistent)\\:\\/\\/(%s)\\/(%s)\\/(%s)" + .formatted(NAME_PATTERN_STR, NAME_PATTERN_STR, NAME_PATTERN_STR)); + + private static final String INVALID_NAME_MSG = "topicName %s must be fully-qualified " + + "in the format 'domain://tenant/namespace/name' where " + + "domain is one of ('persistent', 'non-persistent') and the other components must be " + + "composed of one or more letters, digits, or special characters ('-', '=', ':', or '.')"; + + public PulsarTopic { + Assert.state(TOPIC_NAME_PATTERN.matcher(topicName).matches(), INVALID_NAME_MSG.formatted(topicName)); + Assert.state(numberOfPartitions >= 0, "numberOfPartitions must be >= 0"); + } + + /** + * Convenience method to create a topic builder with the specified topic name. + * @param topicName the name of the topic + * @return the topic builder instance + * @deprecated As of version 1.2.0 topic builder is a registered bean - instead use an + * injected instance where needed + */ + @Deprecated(since = "1.2.0", forRemoval = true) public static PulsarTopicBuilder builder(String topicName) { - return new PulsarTopicBuilder(topicName); + return new PulsarTopicBuilder().name(topicName); } /** @@ -50,28 +82,27 @@ public record PulsarTopic(String topicName, int numberOfPartitions) { * @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."); - + var splitTopic = this.topicName().split("/"); + var type = splitTopic[0].replace(":", ""); + return new TopicComponents(TopicDomain.getEnum(type), splitTopic[2], splitTopic[3], splitTopic[4]); } /** - * Get the fully-qualified name of the topic. + * Get the fully-qualified name of this topic in the format + * {@code domain://tenant/namespace/name} where the components have the following + * defaults when not specified in the original topic name used to build this topic. + * 
+	 * - {@code domain} is one of ('persistent', 'non-persistent') with a default of 'persistent'
+	 * - {@code tenant} has default of 'public'
+	 * - {@code namespace} has default of 'default'
+	 *
* @return the fully-qualified topic name + * @deprecated As of version 1.2.0 topicName must always be fully qualified, use + * {@link #topicName()} instead. */ + @Deprecated(since = "1.2.0", forRemoval = true) public String getFullyQualifiedTopicName() { - TopicComponents components = this.getComponents(); - return components.domain + "://" + components.tenant + "/" + components.namespace + "/" + components.name; + return this.topicName(); } /** 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 index eef03058..a476a820 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopicBuilder.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTopicBuilder.java @@ -16,23 +16,110 @@ package org.springframework.pulsar.core; +import org.apache.pulsar.common.naming.TopicDomain; + +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + /** * Builder class to create {@link PulsarTopic} instances. * * @author Alexander Preuß + * @author Chris Bono */ public class PulsarTopicBuilder { - private final String topicName; + private static final String FQ_TOPIC_NAME_FORMAT = "%s://%s/%s/%s"; + private static final String DEFAULT_TENANT = "public"; + + private static final String DEFAULT_NAMESPACE = "default"; + + private final TopicDomain defaultDomain; + + private final String defaultTenant; + + private final String defaultNamespace; + + @Nullable + private String name; + + @Nullable private int numberOfPartitions; - protected PulsarTopicBuilder(String topicName) { - this.topicName = topicName; + /** + * Create a builder instance that uses the following defaults. 
+	 * - {@code domain -> 'persistent'}
+	 * - {@code tenant -> 'public'}
+	 * - {@code namespace -> 'default'}
+	 *
+ */ + public PulsarTopicBuilder() { + this(TopicDomain.persistent, DEFAULT_TENANT, DEFAULT_NAMESPACE); } /** - * Sets the number of topic partitions. + * Create a builder instance that uses the specified defaults. + * @param defaultDomain domain to use for the topic when not present in the name + * @param defaultTenant tentant to use for the topic when not present in the name + * @param defaultNamespace namespace to use for the topic when not present in the name + */ + public PulsarTopicBuilder(TopicDomain defaultDomain, String defaultTenant, String defaultNamespace) { + Assert.notNull(defaultDomain, "defaultDomain must not be null"); + Assert.hasText(defaultTenant, "defaultTenant must be specified"); + Assert.hasText(defaultNamespace, "defaultNamespace must be specified"); + this.defaultDomain = defaultDomain; + this.defaultTenant = defaultTenant; + this.defaultNamespace = defaultNamespace; + } + + /** + * Get the fully-qualified name of the specified topic in the format + * {@code domain://tenant/namespace/name}. + * @param topicName the topic name to fully qualify + * @return the fully-qualified topic name + */ + public String getFullyQualifiedNameForTopic(String topicName) { + return this.fullyQualifiedName(topicName); + } + + /** + * Set the name of the topic under construction. The following formats are accepted: + * 
+	 * - {@code 'name'}
+	 * - {@code 'tenant/namespace/name'}
+	 * - {@code 'domain://tenant/namespace/name'}
+	 *
When the name is not fully-qualified the missing components are populated + * with the corresponding default configured on the builder. + * @param name the topic name + * @return this builder + */ + public PulsarTopicBuilder name(String name) { + this.name = fullyQualifiedName(name); + return this; + } + + private String fullyQualifiedName(String name) { + Assert.notNull(name, "name must not be null"); + String[] splitTopic = name.split("/"); + if (splitTopic.length == 1) { // e.g. 'my-topic' + return FQ_TOPIC_NAME_FORMAT.formatted(this.defaultDomain, this.defaultTenant, this.defaultNamespace, + splitTopic[0]); + } + if (splitTopic.length == 3) { // e.g. 'public/default/my-topic' + return FQ_TOPIC_NAME_FORMAT.formatted(this.defaultDomain, splitTopic[0], splitTopic[1], splitTopic[2]); + } + if (splitTopic.length == 5) { // e.g. 'persistent://public/default/my-topic' + String type = splitTopic[0].replace(":", ""); + return FQ_TOPIC_NAME_FORMAT.formatted(TopicDomain.getEnum(type), splitTopic[2], splitTopic[3], + splitTopic[4]); + } + throw new IllegalArgumentException("Topic name '" + name + "' must be in one of " + + "the following formats ('name', 'tenant/namespace/name', 'domain://tenant/namespace/name')"); + } + + /** + * Sets the number of topic partitions for the topic under construction. * @param numberOfPartitions the number of topic partitions * @return this builder */ @@ -46,7 +133,7 @@ public class PulsarTopicBuilder { * @return {@link PulsarTopic} */ public PulsarTopic build() { - return new PulsarTopic(this.topicName, this.numberOfPartitions); + return new PulsarTopic(this.name, this.numberOfPartitions); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java index 5eb756aa..53c1be52 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java @@ -89,6 +89,23 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { assertThat(cachedProducerWrapper).isSameAs(producer1); } + @Test + void createProducerWithTopicBuilderAndMultipleCalls() { + var inputTopic = "topic1"; + var fullyQualifiedTopic = "persistent://public/default/topic1"; + var topicBuilder = spy(new PulsarTopicBuilder()); + var producerFactory = producerFactory(pulsarClient, null, null, topicBuilder); + var cacheKey = new ProducerCacheKey<>(schema, fullyQualifiedTopic, null, null); + var producer1 = producerFactory.createProducer(schema, inputTopic); + var producer2 = producerFactory.createProducer(new StringSchema(), inputTopic); + var producer3 = producerFactory.createProducer(new StringSchema(), fullyQualifiedTopic); + assertThat(producer1).isSameAs(producer2).isSameAs(producer3); + CacheProvider, Producer> producerCache = getAssertedProducerCache( + producerFactory, Collections.singletonList(cacheKey)); + Producer cachedProducerWrapper = producerCache.asMap().get(cacheKey); + assertThat(cachedProducerWrapper).isSameAs(producer1); + } + @Test void cachedProducerIsCloseSafeWrapper() throws PulsarClientException { var producerFactory = newProducerFactory(); @@ -169,7 +186,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void factoryDestroyCleansUpCacheAndClosesProducers() { - CachingPulsarProducerFactory producerFactory = producerFactory(pulsarClient, null, null); + CachingPulsarProducerFactory producerFactory = producerFactory(pulsarClient, null, null, null); var actualProducer1 = actualProducer(producerFactory.createProducer(schema, "topic1")); var actualProducer2 = actualProducer(producerFactory.createProducer(schema, "topic2")); var cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null, null); @@ -200,7 +217,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { void createProducerEncountersException() { pulsarClient = spy(pulsarClient); when(this.pulsarClient.newProducer(schema)).thenThrow(new RuntimeException("5150")); - var producerFactory = producerFactory(pulsarClient, null, null); + var producerFactory = producerFactory(pulsarClient, null, null, null); assertThatThrownBy(() -> producerFactory.createProducer(schema, "topic1")).isInstanceOf(RuntimeException.class) .hasMessage("5150"); getAssertedProducerCache(producerFactory, Collections.emptyList()); @@ -230,9 +247,11 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Override protected CachingPulsarProducerFactory producerFactory(PulsarClient pulsarClient, - @Nullable String defaultTopic, @Nullable List> defaultConfigCustomizers) { + @Nullable String defaultTopic, @Nullable List> defaultConfigCustomizers, + @Nullable PulsarTopicBuilder topicBuilder) { var producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2); + producerFactory.setTopicBuilder(topicBuilder); producerFactories.add(producerFactory); return producerFactory; } @@ -331,7 +350,8 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void restartLifecycle() { - var producerFactory = (CachingPulsarProducerFactory) producerFactory(pulsarClient, null, null); + var producerFactory = (CachingPulsarProducerFactory) producerFactory(pulsarClient, null, null, + null); producerFactory.start(); var producer1 = producerFactory.createProducer(schema, "topic1"); var producer2 = producerFactory.createProducer(schema, "topic2"); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java index 1eaf0407..36cb9554 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java @@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.util.Collections; import java.util.List; @@ -267,4 +269,29 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { } + @Nested + class CreateConsumerUsingPulsarTopicBuilder { + + private DefaultPulsarConsumerFactory consumerFactory; + + private PulsarTopicBuilder pulsarTopicBuilder; + + @BeforeEach + void createConsumerFactory() { + pulsarTopicBuilder = spy(new PulsarTopicBuilder()); + consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null); + consumerFactory.setTopicBuilder(pulsarTopicBuilder); + } + + @Test + void withPulsarTopicBuilder() throws PulsarClientException { + try (var consumer = consumerFactory.createConsumer(SCHEMA, Collections.singletonList("topic1"), + "with-pulsar-topic-builder-sub", null, null)) { + assertThat(consumer.getTopic()).isEqualTo("persistent://public/default/topic1"); + verify(pulsarTopicBuilder).getFullyQualifiedNameForTopic("topic1"); + } + } + + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java index fca5f43c..8cdf27df 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java @@ -54,8 +54,12 @@ class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Override protected PulsarProducerFactory producerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, - @Nullable List> defaultConfigCustomizers) { - return new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers); + @Nullable List> defaultConfigCustomizers, + @Nullable PulsarTopicBuilder topicBuilder) { + var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers, + new DefaultTopicResolver()); + producerFactory.setTopicBuilder(topicBuilder); + return producerFactory; } @Nested diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java index 0fd10089..1afec2f4 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java @@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.util.Collections; import java.util.List; @@ -192,6 +194,32 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo } + @Nested + class WithPulsarTopicBuilder { + + private DefaultPulsarReaderFactory pulsarReaderFactory; + + private PulsarTopicBuilder pulsarTopicBuilder; + + @BeforeEach + void createReaderFactory() { + pulsarTopicBuilder = spy(new PulsarTopicBuilder()); + pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, null); + pulsarReaderFactory.setTopicBuilder(pulsarTopicBuilder); + } + + @Test + void topicIsFullyQualified() throws Exception { + var topic = "wptb-reader-topic"; + try (var reader = pulsarReaderFactory.createReader(List.of(topic), MessageId.earliest, Schema.STRING, + Collections.emptyList())) { + assertThat(reader.getTopic()).isEqualTo("persistent://public/default/" + topic); + verify(pulsarTopicBuilder).getFullyQualifiedNameForTopic(topic); + } + } + + } + @Nested @SuppressWarnings("unchecked") class DefaultConfigCustomizerApi { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationIntegrationTests.java index 5d6b75ff..48ecfc85 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationIntegrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationIntegrationTests.java @@ -66,11 +66,11 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer List expectedFullyQualifiedTopicNames = expectedTopics.stream().mapMulti((topic, consumer) -> { if (topic.isPartitioned()) { for (int i = 0; i < topic.numberOfPartitions(); i++) { - consumer.accept(topic.getFullyQualifiedTopicName() + "-partition-" + i); + consumer.accept(topic.topicName() + "-partition-" + i); } } else { - consumer.accept(topic.getFullyQualifiedTopicName()); + consumer.accept(topic.topicName()); } }).toList(); @@ -106,17 +106,17 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Bean PulsarTopic nonPartitionedTopic() { - return PulsarTopic.builder("cmt-non-partitioned-1").build(); + return new PulsarTopicBuilder().name("cmt-non-partitioned-1").build(); } @Bean PulsarTopic nonPartitionedTopic2() { - return PulsarTopic.builder("cmt-non-partitioned-2").build(); + return new PulsarTopicBuilder().name("cmt-non-partitioned-2").build(); } @Bean PulsarTopic partitionedTopic() { - return PulsarTopic.builder("cmt-partitioned-1").numberOfPartitions(4).build(); + return new PulsarTopicBuilder().name("cmt-partitioned-1").numberOfPartitions(4).build(); } } @@ -157,14 +157,14 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Bean PulsarTopic partitionedGreenTopic() { - return PulsarTopic.builder("persistent://%s/partitioned-1".formatted(PUBLIC_GREEN_NAMESPACE)) + return new PulsarTopicBuilder().name("persistent://%s/partitioned-1".formatted(PUBLIC_GREEN_NAMESPACE)) .numberOfPartitions(2) .build(); } @Bean PulsarTopic partitionedBlueTopic() { - return PulsarTopic.builder("persistent://%s/partitioned-1".formatted(PUBLIC_BLUE_NAMESPACE)) + return new PulsarTopicBuilder().name("persistent://%s/partitioned-1".formatted(PUBLIC_BLUE_NAMESPACE)) .numberOfPartitions(2) .build(); } @@ -180,7 +180,7 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Test void topicsExist(@Autowired ObjectProvider expectedTopics) throws Exception { assertThatTopicsExist(expectedTopics.stream().toList()); - PulsarTopic biggerTopic = PulsarTopic.builder("ipc-partitioned-1").numberOfPartitions(4).build(); + PulsarTopic biggerTopic = new PulsarTopicBuilder().name("ipc-partitioned-1").numberOfPartitions(4).build(); pulsarAdministration.createOrModifyTopics(biggerTopic); assertThatTopicsExist(Collections.singletonList(biggerTopic)); } @@ -190,7 +190,7 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Bean PulsarTopic smallerTopic() { - return PulsarTopic.builder("ipc-partitioned-1").numberOfPartitions(1).build(); + return new PulsarTopicBuilder().name("ipc-partitioned-1").numberOfPartitions(1).build(); } } @@ -204,7 +204,7 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Test void topicModificationThrows(@Autowired ObjectProvider expectedTopics) throws Exception { assertThatTopicsExist(expectedTopics.stream().toList()); - PulsarTopic smallerTopic = PulsarTopic.builder("dpc-partitioned-1").numberOfPartitions(4).build(); + PulsarTopic smallerTopic = new PulsarTopicBuilder().name("dpc-partitioned-1").numberOfPartitions(4).build(); assertThatIllegalStateException().isThrownBy(() -> pulsarAdministration.createOrModifyTopics(smallerTopic)) .withMessage( "Topic 'persistent://public/default/dpc-partitioned-1' found w/ 8 partitions but can't shrink to 4 - needs to be deleted first"); @@ -216,7 +216,7 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Bean PulsarTopic biggerTopic() { - return PulsarTopic.builder("dpc-partitioned-1").numberOfPartitions(8).build(); + return new PulsarTopicBuilder().name("dpc-partitioned-1").numberOfPartitions(8).build(); } } @@ -229,7 +229,7 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Test void unpartitionedTopicExists() throws PulsarAdminException { - var topic = PulsarTopic.builder("taet-foo").numberOfPartitions(0).build(); + var topic = new PulsarTopicBuilder().name("taet-foo").numberOfPartitions(0).build(); pulsarAdministration.createOrModifyTopics(topic); assertThatTopicsExist(List.of(topic)); // subsequent call should short circuit and not fail @@ -238,7 +238,7 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Test void partitionedTopicExists() throws PulsarAdminException { - var topic = PulsarTopic.builder("taet-bar").numberOfPartitions(3).build(); + var topic = new PulsarTopicBuilder().name("taet-bar").numberOfPartitions(3).build(); pulsarAdministration.createOrModifyTopics(topic); assertThatTopicsExist(List.of(topic)); // subsequent call should short circuit and not fail @@ -253,8 +253,8 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Test void unpartitionedTopicAlreadyExists() { - var unpartitionedTopic = PulsarTopic.builder("ctt-foo").numberOfPartitions(0).build(); - var partitionedTopic = PulsarTopic.builder("ctt-foo").numberOfPartitions(3).build(); + var unpartitionedTopic = new PulsarTopicBuilder().name("ctt-foo").numberOfPartitions(0).build(); + var partitionedTopic = new PulsarTopicBuilder().name("ctt-foo").numberOfPartitions(3).build(); pulsarAdministration.createOrModifyTopics(unpartitionedTopic); assertThatIllegalStateException() .isThrownBy(() -> pulsarAdministration.createOrModifyTopics(partitionedTopic)) @@ -264,8 +264,8 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer @Test void partitionedTopicAlreadyExists() { - var unpartitionedTopic = PulsarTopic.builder("ctt-bar").numberOfPartitions(0).build(); - var partitionedTopic = PulsarTopic.builder("ctt-bar").numberOfPartitions(3).build(); + var unpartitionedTopic = new PulsarTopicBuilder().name("ctt-bar").numberOfPartitions(0).build(); + var partitionedTopic = new PulsarTopicBuilder().name("ctt-bar").numberOfPartitions(3).build(); pulsarAdministration.createOrModifyTopics(partitionedTopic); assertThatIllegalStateException() .isThrownBy(() -> pulsarAdministration.createOrModifyTopics(unpartitionedTopic)) diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java index 28636b0f..1f36761f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java @@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatNullPointerException; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import java.util.Arrays; @@ -93,20 +94,20 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport } protected PulsarProducerFactory newProducerFactory() { - return producerFactory(pulsarClient, null, null); + return producerFactory(pulsarClient, null, null, null); } protected PulsarProducerFactory newProducerFactoryWithDefaultTopic(String defaultTopic) { - return producerFactory(pulsarClient, defaultTopic, null); + return producerFactory(pulsarClient, defaultTopic, null, null); } private PulsarProducerFactory newProducerFactoryWithDefaultKeys(Set defaultKeys) { - return producerFactory(pulsarClient, null, List.of((pb) -> defaultKeys.forEach(pb::addEncryptionKey))); + return producerFactory(pulsarClient, null, List.of((pb) -> defaultKeys.forEach(pb::addEncryptionKey)), null); } protected PulsarProducerFactory newProducerFactoryWithDefaultConfigCustomizers( List> customizers) { - return producerFactory(pulsarClient, null, customizers); + return producerFactory(pulsarClient, null, customizers, null); } /** @@ -124,10 +125,13 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport * @param defaultTopic the default topic to use for the producers * @param defaultConfigCustomizers the optional list of customizers to apply to the * created producers + * @param topicBuilder the optional topic builder to use for fully qualifying topic + * names * @return a Pulsar producer factory instance to use for the tests */ protected abstract PulsarProducerFactory producerFactory(PulsarClient pulsarClient, - @Nullable String defaultTopic, @Nullable List> defaultConfigCustomizers); + @Nullable String defaultTopic, @Nullable List> defaultConfigCustomizers, + @Nullable PulsarTopicBuilder topicBuilder); @Test @SuppressWarnings("unchecked") @@ -251,4 +255,19 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport } + @Nested + class CreateProducerWithTopicBuilder { + + @Test + void topicIsFullyQualified() throws PulsarClientException { + var topicBuilder = spy(new PulsarTopicBuilder()); + var producerFactory = producerFactory(pulsarClient, null, null, topicBuilder); + try (var producer = producerFactory.createProducer(schema, "topic1")) { + assertThatProducerHasSchemaAndTopic(producer, schema, "persistent://public/default/topic1"); + verify(topicBuilder).getFullyQualifiedNameForTopic("topic1"); + } + } + + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicBuilderTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicBuilderTests.java new file mode 100644 index 00000000..3bf59f0a --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicBuilderTests.java @@ -0,0 +1,83 @@ +/* + * Copyright 2022-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 static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; + +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; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Tests for {@link PulsarTopicBuilder}. + * + * @author Chris Bono + */ +class PulsarTopicBuilderTests { + + private PulsarTopicBuilder builder = new PulsarTopicBuilder(); + + @Test + void whenNumPartitionsNotSpecifiedThenTopicIsNotPartitioned() { + var topicName = "persistent://my-tenant/my-namespace/my-topic"; + var topic = builder.name(topicName).build(); + assertThat(topic.topicName()).isEqualTo(topicName); + assertThat(topic.numberOfPartitions()).isEqualTo(0); + } + + @Test + void whenNumPartitionsSpecifiedThenTopicIsPartitioned() { + var topicName = "persistent://my-tenant/my-namespace/my-topic"; + var topic = builder.name(topicName).numberOfPartitions(5).build(); + assertThat(topic.topicName()).isEqualTo(topicName); + assertThat(topic.numberOfPartitions()).isEqualTo(5); + } + + @ParameterizedTest + @ValueSource(strings = { "persistent://my-namespace/my-topic", "my-namespace/my-topic" }) + void whenNameIsInvalidThenExceptionIsThrown(String invalidName) { + assertThatIllegalArgumentException().isThrownBy(() -> builder.name(invalidName)) + .withMessage("Topic name '" + invalidName + "' must be in one of the following formats " + + "('name', 'tenant/namespace/name', 'domain://tenant/namespace/name')"); + } + + @ParameterizedTest + @MethodSource("nameIsAlwaysFullyQualifiedProvider") + void nameIsAlwaysFullyQualified(PulsarTopicBuilder topicBuilder, String inputTopic, String expectedTopic) { + assertThat(topicBuilder.getFullyQualifiedNameForTopic(inputTopic)).isEqualTo(expectedTopic); + var topic = topicBuilder.name(inputTopic).build(); + assertThat(topic.topicName()).isEqualTo(expectedTopic); + } + + private static Stream nameIsAlwaysFullyQualifiedProvider() { + var defaultBuilder = new PulsarTopicBuilder(); + var customBuilder = new PulsarTopicBuilder(TopicDomain.non_persistent, "my-tenant", "my-namespace"); + return Stream.of(Arguments.of(defaultBuilder, "my-topic", "persistent://public/default/my-topic"), + Arguments.of(defaultBuilder, "foo/bar/my-topic", "persistent://foo/bar/my-topic"), + Arguments.of(defaultBuilder, "non-persistent://foo/bar/my-topic", "non-persistent://foo/bar/my-topic"), + Arguments.of(customBuilder, "my-topic", "non-persistent://my-tenant/my-namespace/my-topic"), + Arguments.of(customBuilder, "foo/bar/my-topic", "non-persistent://foo/bar/my-topic"), + Arguments.of(customBuilder, "persistent://foo/bar/my-topic", "persistent://foo/bar/my-topic")); + } + +} 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 index 27f06c85..8fdaf290 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2023 the original author or authors. + * Copyright 2022-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. @@ -17,63 +17,95 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; - -import java.util.stream.Stream; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; 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; +import org.junit.jupiter.params.provider.ValueSource; /** * Tests for {@link PulsarTopic}. * * @author Alexander Preuß + * @@author Chris Bono */ public class PulsarTopicTests { - @Test - void builderDefaultValues() { - String topicName = "test-default-values"; - PulsarTopicBuilder builder = PulsarTopic.builder(topicName); - PulsarTopic topic = builder.build(); + private static final String FULLY_QUALIFIED_TOPIC = "persistent://public/default/my-topic"; - assertThat(topic.topicName()).isEqualTo(topicName); - assertThat(topic.numberOfPartitions()).isEqualTo(0); + @Test + void whenNegativeNumPartitionsThenExceptionIsThrown() { + assertThatIllegalStateException().isThrownBy(() -> new PulsarTopic(FULLY_QUALIFIED_TOPIC, -1)) + .withMessage("numberOfPartitions must be >= 0"); } @Test - void fakeTestToVerifyReleasePipeline() { + void whenZeroNumPartitionsThenTopicIsNotPartitioned() { + var topic = new PulsarTopic(FULLY_QUALIFIED_TOPIC, 0); + assertThat(topic.numberOfPartitions()).isEqualTo(0); + assertThat(topic.isPartitioned()).isFalse(); + } + @Test + void whenPosititveNumPartitionsThenTopicIsPartitioned() { + var topic = new PulsarTopic(FULLY_QUALIFIED_TOPIC, 2); + assertThat(topic.numberOfPartitions()).isEqualTo(2); + assertThat(topic.isPartitioned()).isTrue(); } @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); + // @formatter:off + @ValueSource(strings = { + "my-domain://public/default/my-topic", + "public/default/my-topic", "my-topic", + "persistent://public/cluster/default/my-topic", + "persistent://publ@c/default/my-topic", + "persistent://public/def@ult/my-topic", + "persistent://public/default/my-t@pic" + }) + // @formatter:on + void whenNameIsInvalidThenExceptionIsThrown(String invalidTopicName) { + var msg = "topicName %s must be fully-qualified in the format".formatted(invalidTopicName); + assertThatIllegalStateException().isThrownBy(() -> new PulsarTopic(invalidTopicName, 0)) + .withMessageStartingWith(msg); } - 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")); + @ParameterizedTest + // @formatter:off + @ValueSource(strings = { + "persistent://public/default/my-topic", + "non-persistent://public/default/my-topic", + "persistent://PUB-=:.7lic/DE-=:.7fault/MY-=:.7topic" + }) + // @formatter:on + void whenNameIsValidThenTopicCreated(String validTopicName) { + var topic = new PulsarTopic(validTopicName, 0); + assertThat(topic.topicName()).isEqualTo(validTopicName); + } + + @Test + void getComponentsReturnsProperComponents() { + var topic = new PulsarTopic("persistent://public/default/my-topic", 0); + var components = topic.getComponents(); + assertThat(components.domain()).isEqualTo(TopicDomain.persistent); + assertThat(components.tenant()).isEqualTo("public"); + assertThat(components.namespace()).isEqualTo("default"); + assertThat(components.name()).isEqualTo("my-topic"); + } + + @Test + @SuppressWarnings({ "deprecation", "removal" }) + void deprecatedBuilderMethodReturnsValidBuilder() { + var fullyQualifiedName = "persistent://public/default/my-topic"; + assertThat(PulsarTopic.builder("my-topic").build().topicName()).isEqualTo(fullyQualifiedName); + } + + @Test + @SuppressWarnings({ "deprecation", "removal" }) + void deprecatedGetFullyQualifiedTopicNameReturnsValidName() { + var topic = new PulsarTopic(FULLY_QUALIFIED_TOPIC, 0); + assertThat(topic.getFullyQualifiedTopicName()).isEqualTo(FULLY_QUALIFIED_TOPIC); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java index 56bc02d7..ae91e7a7 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java @@ -35,6 +35,7 @@ import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.PulsarTopicBuilder; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -96,7 +97,9 @@ abstract class PulsarListenerTestsBase implements PulsarTestContainerSupport { @Bean PulsarTopic partitionedTopic() { - return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build(); + return new PulsarTopicBuilder().name("persistent://public/default/concurrency-on-pl") + .numberOfPartitions(3) + .build(); } } diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 50d983a2..b7618cd4 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -10,6 +10,7 @@ +