Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for default tenant and namespace (#766)

Browse Source 
See #756
This commit is contained in:
Chris Bono
2024-08-12 14:03:32 -05:00
committed by GitHub
parent 96c12de142
commit 6d23378fbb
36 changed files with 1049 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ImperativeAppConfig.java
View File

@@ -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

3
integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ReactiveAppConfig.java
View File

@@ -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

115
integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/DefaultTenantAndNamespaceTests.java Normal file
View File

@@ -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<Integer, Object> expectedMessageFactory) {
var expectedOutput = new ArrayList<String>();
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);
}
}
}
}

110
integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ImperativeAppConfig.java Normal file
View File

@@ -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<Object> pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver,
PulsarTopicBuilder topicBuilder) {
var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, null, null, topicResolver);
producerFactory.setTopicBuilder(topicBuilder);
return producerFactory;
}
@Bean
PulsarConsumerFactory<Object> 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<String> 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);
}
}
}

115
integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ReactiveAppConfig.java Normal file
View File

@@ -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<Object> reactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
PulsarTopicBuilder topicBuilder) {
return DefaultReactivePulsarSenderFactory.builderFor(reactivePulsarClient)
.withTopicBuilder(topicBuilder)
.build();
}
@Bean
ReactivePulsarConsumerFactory<Object> reactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient,
PulsarTopicBuilder topicBuilder) {
var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, List.of());
consumerFactory.setTopicBuilder(topicBuilder);
return consumerFactory;
}
@ReactivePulsarListener(topics = NFQ_TOPIC)
Mono<Void> consumeFromNonFullyQualifiedTopic(String msg) {
LOG.info("++++++CONSUME %s------".formatted(msg));
return Mono.empty();
}
@ReactivePulsarListener(topics = FQ_TOPIC)
Mono<Void> consumeFromFullyQualifiedTopic(String msg) {
LOG.info("++++++CONSUME %s------".formatted(msg));
return Mono.empty();
}
@Bean
ApplicationRunner produceWithDefaultTenantAndNamespace(PulsarAdministration pulsarAdmin,
ReactivePulsarTemplate<String> 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);
}
}
}

1
integration-tests/src/intTest/resources/logback-test.xml
View File

@@ -11,4 +11,5 @@
<logger name="com.github.dockerjava" level="ERROR"/>
<logger name="org.springframework.pulsar.function" level="INFO"/>
<logger name="org.springframework.pulsar.inttest.app" level="INFO"/>
<logger name="org.springframework.pulsar.inttest.config" level="INFO"/>
</configuration>

28
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/default-tenant-namespace.adoc Normal file
View File

@@ -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

13
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc
View File

@@ -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();
}
----

17
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -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#<init>
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

32
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java
View File

@@ -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<T> implements ReactivePulsarCo
@Nullable
private final List<ReactiveMessageConsumerBuilderCustomizer<T>> defaultConfigCustomizers;
@Nullable
private PulsarTopicBuilder topicBuilder;
/**
* Construct an instance.
* @param reactivePulsarClient the reactive client
@@ -53,6 +57,18 @@ public class DefaultReactivePulsarConsumerFactory<T> 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<T> createConsumer(Schema<T> schema) {
return createConsumer(schema, Collections.emptyList());
@@ -61,20 +77,28 @@ public class DefaultReactivePulsarConsumerFactory<T> implements ReactivePulsarCo
@Override
public ReactiveMessageConsumer<T> createConsumer(Schema<T> schema,
List<ReactiveMessageConsumerBuilderCustomizer<T>> customizers) {
ReactiveMessageConsumerBuilder<T> 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<T> 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);
}
}
}

32
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java
View File

@@ -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<T> implements ReactivePulsarRead
@Nullable
private final List<ReactiveMessageReaderBuilderCustomizer<T>> defaultConfigCustomizers;
@Nullable
private PulsarTopicBuilder topicBuilder;
/**
* Construct an instance.
* @param reactivePulsarClient the reactive client
@@ -53,6 +57,18 @@ public class DefaultReactivePulsarReaderFactory<T> 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<T> createReader(Schema<T> schema) {
return createReader(schema, Collections.emptyList());
@@ -61,20 +77,28 @@ public class DefaultReactivePulsarReaderFactory<T> implements ReactivePulsarRead
@Override
public ReactiveMessageReader<T> createReader(Schema<T> schema,
List<ReactiveMessageReaderBuilderCustomizer<T>> customizers) {
ReactiveMessageReaderBuilder<T> 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<T> 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);
}
}
}

35
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
View File

@@ -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<T>
@Nullable
private final List<ReactiveMessageSenderBuilderCustomizer<T>> defaultConfigCustomizers;
@Nullable
private final PulsarTopicBuilder topicBuilder;
private DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient, TopicResolver topicResolver,
@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache, @Nullable String defaultTopic,
@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> defaultConfigCustomizers) {
@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> 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<T>
private ReactiveMessageSender<T> doCreateReactiveMessageSender(Schema<T> schema, @Nullable String topic,
@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> 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<T> sender = this.reactivePulsarClient.messageSender(schema);
@@ -140,6 +146,12 @@ public final class DefaultReactivePulsarSenderFactory<T>
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<T>
private TopicResolver topicResolver = new DefaultTopicResolver();
@Nullable
private PulsarTopicBuilder topicBuilder;
@Nullable
private ReactiveMessageSenderCache messageSenderCache;
@@ -216,6 +231,20 @@ public final class DefaultReactivePulsarSenderFactory<T>
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<T> 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<T>
public DefaultReactivePulsarSenderFactory<T> 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);
}
}

25
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java
View File

@@ -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<String>(
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);
}
}
}

20
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java
View File

@@ -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<String>(
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);
}
}

15
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java
View File

@@ -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.<String>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<String> sender, String expectedTopic) {
assertThatSenderSpecSatisfies(sender,
(senderSpec) -> assertThat(senderSpec).extracting(ReactiveMessageSenderSpec::getTopicName)

5
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTestsBase.java
View File

@@ -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

3
spring-pulsar-sample-apps/sample-failover-custom-router/build.gradle
View File

@@ -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"

3
spring-pulsar-sample-apps/sample-failover-custom-router/src/main/java/com/example/FailoverConsumerApp.java
View File

@@ -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

2
spring-pulsar-sample-apps/sample-imperative-produce-consume/build.gradle
View File

@@ -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')

3
spring-pulsar-sample-apps/sample-imperative-produce-consume/src/main/java/com/example/ImperativeProduceAndConsumeApp.java
View File

@@ -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

18
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java
View File

@@ -52,6 +52,9 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
@Nullable
private final List<ConsumerBuilderCustomizer<T>> defaultConfigCustomizers;
@Nullable
private PulsarTopicBuilder topicBuilder;
/**
* Construct a consumer factory instance.
* @param pulsarClient the client used to consume
@@ -64,6 +67,18 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
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<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics,
@Nullable String subscriptionName, ConsumerBuilderCustomizer<T> customizer) {
@@ -111,6 +126,9 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
}
private void replaceTopicsOnBuilder(ConsumerBuilder<T> builder, Collection<String> topics) {
if (this.topicBuilder != null) {
topics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
}
var builderImpl = (ConsumerBuilderImpl<T>) builder;
builderImpl.getConf().setTopicNames(new HashSet<>(topics));
}

19
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -57,6 +57,9 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
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<T> implements PulsarProducerFactory<T>
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<T> createProducer(Schema<T> schema, @Nullable String topic) {
return doCreateProducer(schema, topic, null, null);
@@ -169,7 +184,9 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
}
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

18
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java
View File

@@ -45,6 +45,9 @@ public class DefaultPulsarReaderFactory<T> implements PulsarReaderFactory<T> {
@Nullable
private final List<ReaderBuilderCustomizer<T>> 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<T> implements PulsarReaderFactory<T> {
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<T> createReader(@Nullable List<String> topics, @Nullable MessageId messageId, Schema<T> schema,
@Nullable List<ReaderBuilderCustomizer<T>> customizers) throws PulsarClientException {
@@ -92,6 +107,9 @@ public class DefaultPulsarReaderFactory<T> implements PulsarReaderFactory<T> {
}
private void replaceTopicsOnBuilder(ReaderBuilder<T> builder, Collection<String> topics) {
if (this.topicBuilder != null) {
topics = topics.stream().map(this.topicBuilder::getFullyQualifiedNameForTopic).toList();
}
var builderImpl = (ReaderBuilderImpl<T>) builder;
builderImpl.getConf().setTopicNames(new HashSet<>(topics));
}

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

@@ -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<String> getMatchingTopicPartitions(PulsarTopic topic, List<String> 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<PulsarTopic> 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<PulsarTopic> 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());
}

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

@@ -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.
* <p>
* The input {@code topicName} must be fully-qualified. As such, it is recommended to use
* the {@link PulsarTopicBuilder} to create instances like this: <pre>{@code
* PulsarTopic topic = new PulsarTopicBuilder().name("my-topic").build();
* }</pre> 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:
*
* <pre>{@code
* PulsarTopic topic = PulsarTopic.builder("topic-name").build();
* }</pre>
*
* @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.
* <pre>
* - {@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'
* </pre>
* @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();
}
/**

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

@@ -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. <pre>
* - {@code domain -> 'persistent'}
* - {@code tenant -> 'public'}
* - {@code namespace -> 'default'}
* </pre>
*/
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:
* <pre>
* - {@code 'name'}
* - {@code 'tenant/namespace/name'}
* - {@code 'domain://tenant/namespace/name'}
* </pre> 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);
}
}

28
spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java
View File

@@ -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<ProducerCacheKey<String>, Producer<String>> producerCache = getAssertedProducerCache(
producerFactory, Collections.singletonList(cacheKey));
Producer<String> 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<String> producerFactory = producerFactory(pulsarClient, null, null);
CachingPulsarProducerFactory<String> 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<String> producerFactory(PulsarClient pulsarClient,
@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers) {
@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> 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<String>) producerFactory(pulsarClient, null, null);
var producerFactory = (CachingPulsarProducerFactory<String>) producerFactory(pulsarClient, null, null,
null);
producerFactory.start();
var producer1 = producerFactory.createProducer(schema, "topic1");
var producer2 = producerFactory.createProducer(schema, "topic2");

27
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java
View File

@@ -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<String> 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");
}
}
}
}

8
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java
View File

@@ -54,8 +54,12 @@ class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
@Override
protected PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic,
@Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers) {
return new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers);
@Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers,
@Nullable PulsarTopicBuilder topicBuilder) {
var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers,
new DefaultTopicResolver());
producerFactory.setTopicBuilder(topicBuilder);
return producerFactory;
}
@Nested

28
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java
View File

@@ -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<String> 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 {

34
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationIntegrationTests.java
View File

@@ -66,11 +66,11 @@ public class PulsarAdministrationIntegrationTests implements PulsarTestContainer
List<String> expectedFullyQualifiedTopicNames = expectedTopics.stream().<String>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<PulsarTopic> 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<PulsarTopic> 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))

29
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java
View File

@@ -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<String> newProducerFactory() {
return producerFactory(pulsarClient, null, null);
return producerFactory(pulsarClient, null, null, null);
}
protected PulsarProducerFactory<String> newProducerFactoryWithDefaultTopic(String defaultTopic) {
return producerFactory(pulsarClient, defaultTopic, null);
return producerFactory(pulsarClient, defaultTopic, null, null);
}
private PulsarProducerFactory<String> newProducerFactoryWithDefaultKeys(Set<String> 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<String> newProducerFactoryWithDefaultConfigCustomizers(
List<ProducerBuilderCustomizer<String>> 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<String> producerFactory(PulsarClient pulsarClient,
@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers);
@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> 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");
}
}
}
}

83
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTopicBuilderTests.java Normal file
View File

@@ -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<Arguments> 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"));
}
}

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

@@ -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<Arguments> topicComponentsProvider() {
return Stream.of(
Arguments.of(PulsarTopic.builder("topic-1").build(), TopicDomain.persistent, "public", "default",
"topic-1"),
Arguments.of(PulsarTopic.builder("public/default/topic-2").build(), TopicDomain.persistent, "public",
"default", "topic-2"),
Arguments.of(PulsarTopic.builder("persistent://public/default/topic-3").build(), TopicDomain.persistent,
"public", "default", "topic-3"),
Arguments.of(PulsarTopic.builder("public/my-namespace/topic-4").build(), TopicDomain.persistent,
"public", "my-namespace", "topic-4"),
Arguments.of(PulsarTopic.builder("my-tenant/my-namespace/topic-5").build(), TopicDomain.persistent,
"my-tenant", "my-namespace", "topic-5"),
Arguments.of(PulsarTopic.builder("non-persistent://public/my-namespace/topic-6").build(),
TopicDomain.non_persistent, "public", "my-namespace", "topic-6"),
Arguments.of(PulsarTopic.builder("non-persistent://my-tenant/my-namespace/topic-7").build(),
TopicDomain.non_persistent, "my-tenant", "my-namespace", "topic-7"));
@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);
}
}

5
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTestsBase.java
View File

@@ -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();
}
}

1
src/checkstyle/checkstyle-suppressions.xml
View File

@@ -10,6 +10,7 @@
<suppress files="Proto" checks=".*"/>
<suppress files="ReactiveSpringPulsarBootApp" checks="HideUtilityClassConstructor"/>
<suppress files="SamplePulsarApplicationTests" checks="HideUtilityClassConstructor" />
<suppress files="DefaultTenantAndNamespaceTests" checks="HideUtilityClassConstructor" />
<suppress files="[\\/]spring-pulsar-docs[\\/]" checks="JavadocPackage|JavadocType|JavadocVariable|SpringDeprecatedCheck" />
<suppress files="[\\/]spring-pulsar-docs[\\/]" checks="SpringJavadoc" message="\@since" />
<suppress files="[\\/]spring-pulsar-docs[\\/].*jooq" checks="AvoidStaticImport" />