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

Producer access mode related changes

Browse Source
This commit is contained in:
Soby Chacko
2022-07-06 19:30:11 -04:00
parent 6057582cd2
commit 25dc936ec6
2 changed files with 16 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
README.adoc
View File

@@ -74,6 +74,7 @@ spring.pulsar.producer.batchingEnabled
spring.pulsar.producer.chunkingEnabled
spring.pulsar.producer.compressionType
spring.pulsar.producer.initialSubscriptionName
spring.pulsar.producer.accessMode
```
#### Pulsar Consumer
@@ -217,7 +218,7 @@ It is a topic that is partitioned and for this sample we assume that the topic i
public class PulsarBootPartitioned {
public static void main(String[] args) {
SpringApplication.run(PulsarBootPartitioned.class, args);
SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition");
}
@Bean
@@ -266,11 +267,12 @@ public class PulsarBootPartitioned {
A few things require explanation in the application above.
We are publishing to a partitioned topic and we would like to publish some data segment to a specific partition.
If you leave it to Pulsar's default, it follows a round-robin mode of partition assignments and we would like to override that.
If you leave it to Pulsar's default, it follows a round-robin mode of partition assignments, and we would like to override that.
In order to do that, we are providing a message router object with the send method.
Look at the three message routers implemented.
`FooRouter` always sends data to partition `0`, `BarRouter` to partition `1` and `BuzzRouter` to partition `2`.
Also note that, we are now using the `sendAsync` method of `PulsarTemplate` that returns a `CompletableFuture`.
When running the application, we also need to set the `messageRoutingMode` on the producer to `CustomPartition` (`spring.pulsar.producer.messageRoutingMode`).
On the consumer side, we are using a `PulsarListener` with the exclusive subscription type.
This means that data from all the partitions will end up in the same consumer and there is no ordering guarantee.

12
spring-pulsar-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -28,6 +28,7 @@ import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.HashingScheme;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.ProducerAccessMode;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
@@ -380,6 +381,8 @@ public class PulsarProperties {
private String initialSubscriptionName;
private ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared;
public String getTopicName() {
return topicName;
}
@@ -500,6 +503,14 @@ public class PulsarProperties {
this.initialSubscriptionName = initialSubscriptionName;
}
public ProducerAccessMode getProducerAccessMode() {
return producerAccessMode;
}
public void setProducerAccessMode(ProducerAccessMode producerAccessMode) {
this.producerAccessMode = producerAccessMode;
}
public Map<String, Object> buildProperties() {
PulsarProperties.Properties properties = new Properties();
@@ -520,6 +531,7 @@ public class PulsarProperties {
map.from(this::isChunkingEnabled).to(properties.in("chunkingEnabled"));
map.from(this::getCompressionType).to(properties.in("compressionType"));
map.from(this::getInitialSubscriptionName).to(properties.in("initialSubscriptionName"));
map.from(this::getProducerAccessMode).to(properties.in("accessMode"));
return properties;
}