Skip to content

Commit 6d23378

Browse files
onobconobc
authored
Add support for default tenant and namespace (#766)
See #756
1 parent 96c12de commit 6d23378

File tree

36 files changed

+1047
-127
lines changed

36 files changed

+1047
-127
lines changed

‎integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ImperativeAppConfig.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.springframework.pulsar.annotation.PulsarListener;
2828
import org.springframework.pulsar.core.PulsarTemplate;
2929
import org.springframework.pulsar.core.PulsarTopic;
30+
import org.springframework.pulsar.core.PulsarTopicBuilder;
3031

3132
@SpringBootConfiguration
3233
@EnableAutoConfiguration
@@ -39,7 +40,7 @@ class ImperativeAppConfig {
3940

4041
@Bean
4142
PulsarTopic pulsarTestTopic() {
42-
return PulsarTopic.builder(TOPIC).numberOfPartitions(1).build();
43+
return new PulsarTopicBuilder().name(TOPIC).numberOfPartitions(1).build();
4344
}
4445

4546
@Bean

‎integration-tests/src/intTest/java/org/springframework/pulsar/inttest/app/ReactiveAppConfig.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.springframework.context.annotation.Bean;
2727
import org.springframework.context.annotation.Profile;
2828
import org.springframework.pulsar.core.PulsarTopic;
29+
import org.springframework.pulsar.core.PulsarTopicBuilder;
2930
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
3031
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
3132

@@ -43,7 +44,7 @@ class ReactiveAppConfig {
4344

4445
@Bean
4546
PulsarTopic pulsarTestTopic() {
46-
return PulsarTopic.builder(TOPIC).numberOfPartitions(1).build();
47+
return new PulsarTopicBuilder().name(TOPIC).numberOfPartitions(1).build();
4748
}
4849

4950
@Bean

‎integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/DefaultTenantAndNamespaceTests.java

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* Copyright 2012-2023 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.pulsar.inttest.config;
18+
19+
import static org.assertj.core.api.Assertions.assertThat;
20+
21+
import java.time.Duration;
22+
import java.util.ArrayList;
23+
import java.util.function.Function;
24+
import java.util.stream.IntStream;
25+
26+
import org.awaitility.Awaitility;
27+
import org.junit.jupiter.api.Nested;
28+
import org.junit.jupiter.api.Test;
29+
import org.junit.jupiter.api.extension.ExtendWith;
30+
import org.testcontainers.containers.PulsarContainer;
31+
import org.testcontainers.junit.jupiter.Container;
32+
import org.testcontainers.junit.jupiter.Testcontainers;
33+
34+
import org.springframework.beans.factory.annotation.Autowired;
35+
import org.springframework.boot.test.context.SpringBootTest;
36+
import org.springframework.boot.test.system.CapturedOutput;
37+
import org.springframework.boot.test.system.OutputCaptureExtension;
38+
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
39+
import org.springframework.pulsar.core.PulsarAdministration;
40+
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
41+
import org.springframework.test.context.ActiveProfiles;
42+
43+
@Testcontainers(disabledWithoutDocker = true)
44+
@ExtendWith(OutputCaptureExtension.class)
45+
class DefaultTenantAndNamespaceTests {
46+
47+
@SuppressWarnings("unused")
48+
@Container
49+
@ServiceConnection
50+
static PulsarContainer PULSAR_CONTAINER = new PulsarContainer(PulsarTestContainerSupport.getPulsarImage());
51+
52+
@Nested
53+
@SpringBootTest(classes = ImperativeAppConfig.class)
54+
@ExtendWith(OutputCaptureExtension.class)
55+
@ActiveProfiles("inttest.pulsar.imperative")
56+
class WithImperativeApp {
57+
58+
@Test
59+
void produceConsumeWithDefaultTenantNamespace(CapturedOutput output,
60+
@Autowired PulsarAdministration pulsarAdmin) {
61+
TestVerifyUtils.verifyProduceConsume(output, 10, (i) -> ImperativeAppConfig.MSG_PREFIX + i);
62+
TestVerifyUtils.verifyTopicsLocatedInTenantAndNamespace(pulsarAdmin, ImperativeAppConfig.TENANT,
63+
ImperativeAppConfig.NAMESPACE, ImperativeAppConfig.NFQ_TOPIC);
64+
}
65+
66+
}
67+
68+
@Nested
69+
@SpringBootTest(classes = ReactiveAppConfig.class)
70+
@ExtendWith(OutputCaptureExtension.class)
71+
@ActiveProfiles("inttest.pulsar.reactive")
72+
class WithReactiveApp {
73+
74+
@Test
75+
void produceConsumeWithDefaultTenantNamespace(CapturedOutput output,
76+
@Autowired PulsarAdministration pulsarAdmin) {
77+
TestVerifyUtils.verifyProduceConsume(output, 10, (i) -> ReactiveAppConfig.MSG_PREFIX + i);
78+
TestVerifyUtils.verifyTopicsLocatedInTenantAndNamespace(pulsarAdmin, ReactiveAppConfig.TENANT,
79+
ReactiveAppConfig.NAMESPACE, ReactiveAppConfig.NFQ_TOPIC);
80+
}
81+
82+
}
83+
84+
private static class TestVerifyUtils {
85+
86+
static void verifyProduceConsume(CapturedOutput output, int numExpectedMessages,
87+
Function<Integer, Object> expectedMessageFactory) {
88+
var expectedOutput = new ArrayList<String>();
89+
IntStream.range(0, numExpectedMessages).forEachOrdered((i) -> {
90+
var expectedMsg = expectedMessageFactory.apply(i);
91+
expectedOutput.add("++++++PRODUCE %s------".formatted(expectedMsg));
92+
expectedOutput.add("++++++CONSUME %s------".formatted(expectedMsg));
93+
});
94+
Awaitility.waitAtMost(Duration.ofSeconds(15))
95+
.untilAsserted(() -> assertThat(output).contains(expectedOutput));
96+
}
97+
98+
static void verifyTopicsLocatedInTenantAndNamespace(PulsarAdministration pulsarAdmin, String tenant,
99+
String namespace, String topic) {
100+
// verify topics created in expected tenant/namespace and not in
101+
// public/default
102+
try (var admin = pulsarAdmin.createAdminClient()) {
103+
var fqTopic = "persistent://%s/%s/%s".formatted(tenant, namespace, topic);
104+
assertThat(admin.topics().getList("%s/%s".formatted(tenant, namespace))).containsExactly(fqTopic);
105+
assertThat(admin.topics().getList("public/default"))
106+
.noneSatisfy(t -> assertThat(t).doesNotEndWith("/" + topic));
107+
}
108+
catch (Exception ex) {
109+
throw new RuntimeException(ex);
110+
}
111+
}
112+
113+
}
114+
115+
}

‎integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ImperativeAppConfig.java

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
/*
2+
* Copyright 2023-2024 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.pulsar.inttest.config;
18+
19+
import java.util.Set;
20+
21+
import org.apache.commons.logging.Log;
22+
import org.apache.commons.logging.LogFactory;
23+
import org.apache.pulsar.client.api.PulsarClient;
24+
import org.apache.pulsar.common.naming.TopicDomain;
25+
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
26+
27+
import org.springframework.boot.ApplicationRunner;
28+
import org.springframework.boot.SpringBootConfiguration;
29+
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
30+
import org.springframework.context.annotation.Bean;
31+
import org.springframework.context.annotation.Profile;
32+
import org.springframework.pulsar.annotation.PulsarListener;
33+
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
34+
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
35+
import org.springframework.pulsar.core.PulsarAdministration;
36+
import org.springframework.pulsar.core.PulsarConsumerFactory;
37+
import org.springframework.pulsar.core.PulsarProducerFactory;
38+
import org.springframework.pulsar.core.PulsarTemplate;
39+
import org.springframework.pulsar.core.PulsarTopicBuilder;
40+
import org.springframework.pulsar.core.TopicResolver;
41+
42+
@SpringBootConfiguration
43+
@EnableAutoConfiguration
44+
@Profile("inttest.pulsar.imperative")
45+
class ImperativeAppConfig {
46+
47+
private static final Log LOG = LogFactory.getLog(ImperativeAppConfig.class);
48+
static final String TENANT = "my-tenant-i";
49+
static final String NAMESPACE = "my-namespace-i";
50+
static final String NFQ_TOPIC = "dtant-topic-i";
51+
static final String FQ_TOPIC = "persistent://my-tenant-i/my-namespace-i/dtant-topic-i";
52+
static final String MSG_PREFIX = "DefaultTenantNamespace-i:";
53+
54+
@Bean
55+
PulsarTopicBuilder topicBuilder() {
56+
return new PulsarTopicBuilder(TopicDomain.persistent, TENANT, NAMESPACE);
57+
}
58+
59+
@Bean
60+
PulsarProducerFactory<Object> pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver,
61+
PulsarTopicBuilder topicBuilder) {
62+
var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, null, null, topicResolver);
63+
producerFactory.setTopicBuilder(topicBuilder);
64+
return producerFactory;
65+
}
66+
67+
@Bean
68+
PulsarConsumerFactory<Object> pulsarConsumerFactory(PulsarClient pulsarClient, PulsarTopicBuilder topicBuilder) {
69+
var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null);
70+
consumerFactory.setTopicBuilder(topicBuilder);
71+
return consumerFactory;
72+
}
73+
74+
@PulsarListener(topics = NFQ_TOPIC)
75+
void consumeFromNonFullyQualifiedTopic(String msg) {
76+
LOG.info("++++++CONSUME %s------".formatted(msg));
77+
}
78+
79+
@PulsarListener(topics = FQ_TOPIC)
80+
void consumeFromFullyQualifiedTopic(String msg) {
81+
LOG.info("++++++CONSUME %s------".formatted(msg));
82+
}
83+
84+
@Bean
85+
ApplicationRunner produceWithDefaultTenantAndNamespace(PulsarAdministration pulsarAdmin,
86+
PulsarTemplate<String> template) {
87+
createTenantAndNamespace(pulsarAdmin);
88+
return (args) -> {
89+
for (int i = 0; i < 10; i++) {
90+
var msg = MSG_PREFIX + i;
91+
template.send((i < 5) ? FQ_TOPIC : NFQ_TOPIC, msg);
92+
LOG.info("++++++PRODUCE %s------".formatted(msg));
93+
}
94+
};
95+
}
96+
97+
private void createTenantAndNamespace(PulsarAdministration pulsarAdmin) {
98+
try (var admin = pulsarAdmin.createAdminClient()) {
99+
admin.tenants()
100+
.createTenant(TENANT, TenantInfoImpl.builder().allowedClusters(Set.of("standalone")).build());
101+
LOG.info("Created tenant -> %s".formatted(admin.tenants().getTenantInfo(TENANT)));
102+
admin.namespaces().createNamespace("%s/%s".formatted(TENANT, NAMESPACE));
103+
LOG.info("Created namespace -> %s".formatted(admin.namespaces().getNamespaces(TENANT)));
104+
}
105+
catch (Exception ex) {
106+
throw new RuntimeException(ex);
107+
}
108+
}
109+
110+
}

‎integration-tests/src/intTest/java/org/springframework/pulsar/inttest/config/ReactiveAppConfig.java

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
/*
2+
* Copyright 2023-2024 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.pulsar.inttest.config;
18+
19+
import java.util.List;
20+
import java.util.Set;
21+
22+
import org.apache.commons.logging.Log;
23+
import org.apache.commons.logging.LogFactory;
24+
import org.apache.pulsar.common.naming.TopicDomain;
25+
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
26+
import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
27+
28+
import org.springframework.boot.ApplicationRunner;
29+
import org.springframework.boot.SpringBootConfiguration;
30+
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
31+
import org.springframework.context.annotation.Bean;
32+
import org.springframework.context.annotation.Profile;
33+
import org.springframework.pulsar.core.PulsarAdministration;
34+
import org.springframework.pulsar.core.PulsarTopicBuilder;
35+
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
36+
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
37+
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
38+
import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
39+
import org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory;
40+
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
41+
42+
import reactor.core.publisher.Mono;
43+
44+
@SpringBootConfiguration
45+
@EnableAutoConfiguration
46+
@Profile("inttest.pulsar.reactive")
47+
class ReactiveAppConfig {
48+
49+
private static final Log LOG = LogFactory.getLog(ReactiveAppConfig.class);
50+
static final String TENANT = "my-tenant-r";
51+
static final String NAMESPACE = "my-namespace-r";
52+
static final String NFQ_TOPIC = "dtant-topic-r";
53+
static final String FQ_TOPIC = "persistent://my-tenant-r/my-namespace-r/dtant-topic-r";
54+
static final String MSG_PREFIX = "DefaultTenantNamespace-r:";
55+
56+
@Bean
57+
PulsarTopicBuilder topicBuilder() {
58+
return new PulsarTopicBuilder(TopicDomain.persistent, TENANT, NAMESPACE);
59+
}
60+
61+
@Bean
62+
ReactivePulsarSenderFactory<Object> reactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
63+
PulsarTopicBuilder topicBuilder) {
64+
return DefaultReactivePulsarSenderFactory.builderFor(reactivePulsarClient)
65+
.withTopicBuilder(topicBuilder)
66+
.build();
67+
}
68+
69+
@Bean
70+
ReactivePulsarConsumerFactory<Object> reactivePulsarConsumerFactory(ReactivePulsarClient reactivePulsarClient,
71+
PulsarTopicBuilder topicBuilder) {
72+
var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, List.of());
73+
consumerFactory.setTopicBuilder(topicBuilder);
74+
return consumerFactory;
75+
}
76+
77+
@ReactivePulsarListener(topics = NFQ_TOPIC)
78+
Mono<Void> consumeFromNonFullyQualifiedTopic(String msg) {
79+
LOG.info("++++++CONSUME %s------".formatted(msg));
80+
return Mono.empty();
81+
}
82+
83+
@ReactivePulsarListener(topics = FQ_TOPIC)
84+
Mono<Void> consumeFromFullyQualifiedTopic(String msg) {
85+
LOG.info("++++++CONSUME %s------".formatted(msg));
86+
return Mono.empty();
87+
}
88+
89+
@Bean
90+
ApplicationRunner produceWithDefaultTenantAndNamespace(PulsarAdministration pulsarAdmin,
91+
ReactivePulsarTemplate<String> template) {
92+
createTenantAndNamespace(pulsarAdmin);
93+
return (args) -> {
94+
for (int i = 0; i < 10; i++) {
95+
var msg = MSG_PREFIX + i;
96+
template.send((i < 5) ? FQ_TOPIC : NFQ_TOPIC, msg).subscribe();
97+
LOG.info("++++++PRODUCE %s------".formatted(msg));
98+
}
99+
};
100+
}
101+
102+
private void createTenantAndNamespace(PulsarAdministration pulsarAdmin) {
103+
try (var admin = pulsarAdmin.createAdminClient()) {
104+
admin.tenants()
105+
.createTenant(TENANT, TenantInfoImpl.builder().allowedClusters(Set.of("standalone")).build());
106+
LOG.info("Created tenant -> %s".formatted(admin.tenants().getTenantInfo(TENANT)));
107+
admin.namespaces().createNamespace("%s/%s".formatted(TENANT, NAMESPACE));
108+
LOG.info("Created namespace -> %s".formatted(admin.namespaces().getNamespaces(TENANT)));
109+
}
110+
catch (Exception ex) {
111+
throw new RuntimeException(ex);
112+
}
113+
}
114+
115+
}

‎integration-tests/src/intTest/resources/logback-test.xml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,4 +11,5 @@
1111
<logger name="com.github.dockerjava" level="ERROR"/>
1212
<logger name="org.springframework.pulsar.function" level="INFO"/>
1313
<logger name="org.springframework.pulsar.inttest.app" level="INFO"/>
14+
<logger name="org.springframework.pulsar.inttest.config" level="INFO"/>
1415
</configuration>

0 commit comments

Comments
 (0)