SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use PulsarTopicBuilder in Pulsar binder (#2999)

Browse Source 
Spring for Apache Pulsar introduced support for default tenant and namespace
for Pulsar topics in 6d23378fbb.
This ensures that all topic names are fully-qualified (using the default
tenant and namespace when not fully-qualified).
This commit is contained in:
Chris Bono
2024-09-03 18:20:18 -05:00
committed by GitHub
parent fe1392efb0
commit 64dd18cc8e
5 changed files with 41 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/config/PulsarBinderConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023-2023 the original author or authors.
* 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.
@@ -29,6 +29,7 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.support.header.JacksonUtils;
import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper;
@@ -39,6 +40,7 @@ import org.springframework.pulsar.support.header.ToStringPulsarHeaderMapper;
* Pulsar binder {@link Configuration}.
*
* @author Soby Chacko
* @author Chris Bono
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingBean(Binder.class)
@@ -48,8 +50,9 @@ public class PulsarBinderConfiguration {
@Bean
public PulsarTopicProvisioner pulsarTopicProvisioner(PulsarAdministration pulsarAdministration,
PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties) {
return new PulsarTopicProvisioner(pulsarAdministration, pulsarBinderConfigurationProperties);
PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties,
PulsarTopicBuilder topicBuilder) {
return new PulsarTopicProvisioner(pulsarAdministration, pulsarBinderConfigurationProperties, topicBuilder);
}
@Bean

18
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/provisioning/PulsarTopicProvisioner.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.
@@ -27,12 +27,13 @@ import org.springframework.cloud.stream.provisioning.ProvisioningException;
import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarTopic;
import org.springframework.pulsar.core.PulsarTopicBuilder;
/**
* Pulsar topic provisioner.
*
* @author Soby Chacko
* @author Chris Bono
*/
public class PulsarTopicProvisioner implements
ProvisioningProvider<ExtendedConsumerProperties<PulsarConsumerProperties>, ExtendedProducerProperties<PulsarProducerProperties>> {
@@ -41,10 +42,19 @@ public class PulsarTopicProvisioner implements
private final PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties;
private final PulsarTopicBuilder topicBuilder;
public PulsarTopicProvisioner(PulsarAdministration pulsarAdministration,
PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties) {
this(pulsarAdministration, pulsarBinderConfigurationProperties, new PulsarTopicBuilder());
}
public PulsarTopicProvisioner(PulsarAdministration pulsarAdministration,
PulsarBinderConfigurationProperties pulsarBinderConfigurationProperties,
PulsarTopicBuilder topicBuilder) {
this.pulsarAdministration = pulsarAdministration;
this.pulsarBinderConfigurationProperties = pulsarBinderConfigurationProperties;
this.topicBuilder = topicBuilder;
}
@Override
@@ -53,7 +63,7 @@ public class PulsarTopicProvisioner implements
throws ProvisioningException {
Integer partitionCountFromBinding = pulsarProducerProperties.getExtension().getPartitionCount();
var partitionCount = getPartitionCount(partitionCountFromBinding);
var pulsarTopic = PulsarTopic.builder(name).numberOfPartitions(partitionCount).build();
var pulsarTopic = this.topicBuilder.name(name).numberOfPartitions(partitionCount).build();
this.pulsarAdministration.createOrModifyTopics(pulsarTopic);
return new PulsarDestination(pulsarTopic.topicName(), pulsarTopic.numberOfPartitions());
}
@@ -72,7 +82,7 @@ public class PulsarTopicProvisioner implements
throws ProvisioningException {
var partitionCountFromBinding = pulsarConsumerProperties.getExtension().getPartitionCount();
var partitionCount = getPartitionCount(partitionCountFromBinding);
var pulsarTopic = PulsarTopic.builder(name).numberOfPartitions(partitionCount).build();
var pulsarTopic = this.topicBuilder.name(name).numberOfPartitions(partitionCount).build();
this.pulsarAdministration.createOrModifyTopics(pulsarTopic);
return new PulsarDestination(pulsarTopic.topicName(), pulsarTopic.numberOfPartitions());
}

6
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023-2023 the original author or authors.
* 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.
@@ -55,6 +55,7 @@ import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.PulsarTopicBuilder;
import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
@@ -101,7 +102,8 @@ class PulsarBinderTests extends
protected PulsarTestBinder getBinder() {
var pulsarAdministration = new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl());
var configProps = new PulsarBinderConfigurationProperties();
var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps);
var topicBuilder = new PulsarTopicBuilder();
var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps, topicBuilder);
var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient);
var pulsarTemplate = new PulsarTemplate<>(producerFactory);
var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,

33
binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarTopicProvisionerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2023-2023 the original author or authors.
* 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.
@@ -50,17 +50,7 @@ class PulsarTopicProvisionerTests {
new PulsarProducerProperties());
ProducerDestination producerDestination = pulsarTopicProvisioner.provisionProducerDestination("foo",
properties);
verifyAndAssert(pulsarAdministration, producerDestination.getName(), "foo", 0);
}
private static void verifyAndAssert(PulsarAdministration pulsarAdministration, String actualProducerDestination,
String expectedProducerDestination, int expectedPartitionCount) {
ArgumentCaptor<PulsarTopic> pulsarTopicArgumentCaptor = ArgumentCaptor.forClass(PulsarTopic.class);
verify(pulsarAdministration, times(1)).createOrModifyTopics(pulsarTopicArgumentCaptor.capture());
assertThat(actualProducerDestination).isEqualTo(expectedProducerDestination);
PulsarTopic pulsarTopic = pulsarTopicArgumentCaptor.getValue();
assertThat(pulsarTopic.topicName()).isEqualTo(expectedProducerDestination);
assertThat(pulsarTopic.numberOfPartitions()).isEqualTo(expectedPartitionCount);
verifyAndAssert(pulsarAdministration, producerDestination.getName(), "persistent://public/default/foo", 0);
}
@Test
@@ -73,7 +63,7 @@ class PulsarTopicProvisionerTests {
new PulsarConsumerProperties());
ConsumerDestination consumerDestination = pulsarTopicProvisioner.provisionConsumerDestination("bar", "",
properties);
verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "bar", 0);
verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "persistent://public/default/bar", 0);
}
@Test
@@ -87,7 +77,7 @@ class PulsarTopicProvisionerTests {
new PulsarProducerProperties());
ProducerDestination producerDestination = pulsarTopicProvisioner.provisionProducerDestination("foo",
properties);
verifyAndAssert(pulsarAdministration, producerDestination.getName(), "foo", 4);
verifyAndAssert(pulsarAdministration, producerDestination.getName(), "persistent://public/default/foo", 4);
}
@Test
@@ -101,7 +91,7 @@ class PulsarTopicProvisionerTests {
properties.getExtension().setPartitionCount(4);
ProducerDestination producerDestination = pulsarTopicProvisioner.provisionProducerDestination("foo",
properties);
verifyAndAssert(pulsarAdministration, producerDestination.getName(), "foo", 4);
verifyAndAssert(pulsarAdministration, producerDestination.getName(), "persistent://public/default/foo", 4);
}
@Test
@@ -115,7 +105,7 @@ class PulsarTopicProvisionerTests {
new PulsarConsumerProperties());
ConsumerDestination consumerDestination = pulsarTopicProvisioner.provisionConsumerDestination("bar", "",
properties);
verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "bar", 4);
verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "persistent://public/default/bar", 4);
}
@Test
@@ -130,7 +120,16 @@ class PulsarTopicProvisionerTests {
pulsarConsumerProperties);
ConsumerDestination consumerDestination = pulsarTopicProvisioner.provisionConsumerDestination("bar", "",
properties);
verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "bar", 4);
verifyAndAssert(pulsarAdministration, consumerDestination.getName(), "persistent://public/default/bar", 4);
}
private static void verifyAndAssert(PulsarAdministration pulsarAdministration, String actualProducerDestination,
String expectedProducerDestination, int expectedPartitionCount) {
ArgumentCaptor<PulsarTopic> pulsarTopicArgumentCaptor = ArgumentCaptor.forClass(PulsarTopic.class);
verify(pulsarAdministration, times(1)).createOrModifyTopics(pulsarTopicArgumentCaptor.capture());
assertThat(actualProducerDestination).isEqualTo(expectedProducerDestination);
PulsarTopic pulsarTopic = pulsarTopicArgumentCaptor.getValue();
assertThat(pulsarTopic.topicName()).isEqualTo(expectedProducerDestination);
assertThat(pulsarTopic.numberOfPartitions()).isEqualTo(expectedPartitionCount);
}
}

2
bom/spring-cloud-starter-parent/pom.xml
View File

@@ -6,7 +6,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.4.0-M1</version>
<version>3.4.0-M2</version>
<relativePath/>
</parent>
<groupId>org.springframework.cloud</groupId>