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

Polish reference docs

Browse Source 
- Remove hardcoded code where possible
- Link to auto-generated config props rather than hardcoded tables
- Simplify some wording
This commit is contained in:
Chris Bono
2022-08-11 01:33:48 -05:00
committed by Soby Chacko
parent ade2c74482
commit b145c3ff82
10 changed files with 136 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/asciidoc/AsciidoctorConventionsPlugin.java
View File

@@ -133,6 +133,7 @@ public class AsciidoctorConventionsPlugin implements Plugin<Project> {
attributes.put("attribute-missing", "warn");
attributes.put("github-tag", determineGitHubTag(project));
attributes.put("revnumber", null);
attributes.put("allow-uri-read", ""); // allow remote includes
asciidoctorTask.attributes(attributes);
}

5
buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/ConfigurationPropertiesPlugin.java
View File

@@ -79,11 +79,6 @@ public class ConfigurationPropertiesPlugin implements Plugin<Project> {
private void addConfigurationProcessorDependency(Project project) {
Configuration annotationProcessors = project.getConfigurations()
.getByName(JavaPlugin.ANNOTATION_PROCESSOR_CONFIGURATION_NAME);
// Object bootVersion = project.findProperty("springBootVersion");
// if (bootVersion == null) {
// bootVersion = project.findProperty("spring-boot.version");
// }
// Assert.notNull(bootVersion, "Unable to determine Spring Boot version");
annotationProcessors.getDependencies().add(project.getDependencies().create(
"org.springframework.boot:spring-boot-configuration-processor"));
}

12
buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java
View File

@@ -63,12 +63,12 @@ public class DocumentConfigurationProperties extends DefaultTask {
@TaskAction
void documentConfigurationProperties() throws IOException {
Snippets snippets = new Snippets(this.configurationPropertyMetadata);
snippets.add("application-properties.pulsar", "Pulsar Properties", this::pulsarPrefixes);
snippets.add("application-properties.pulsar-client", "Pulsar Client Properties", (c) -> c.accept("spring.pulsar.client"));
snippets.add("application-properties.pulsar-producer", "Pulsar Producer Properties", (c) -> c.accept("spring.pulsar.producer"));
snippets.add("application-properties.pulsar-consumer", "Pulsar Consumer Properties", (c) -> {
c.accept("spring.pulsar.consumer");
c.accept("spring.pulsar.listener");
});
snippets.writeTo(this.outputDir.toPath());
}
private void pulsarPrefixes(Snippet.Config config) {
config.accept("spring.pulsar");
}
}

6
spring-pulsar-docs/src/main/asciidoc/application-properties.adoc
View File

@@ -10,4 +10,8 @@ TIP: Spring Boot provides various conversion mechanism with advanced value forma
NOTE: Property contributions can come from additional jar files on your classpath, so you should not consider this an exhaustive list.
Also, you can define your own properties.
include::application-properties/pulsar.adoc[]
include::application-properties/pulsar-client.adoc[]
include::application-properties/pulsar-producer.adoc[]
include::application-properties/pulsar-consumer.adoc[]

2
spring-pulsar-docs/src/main/asciidoc/index.adoc
View File

@@ -5,7 +5,7 @@
:numbered:
:icons: font
:hide-uri-scheme:
Soby Chacko
Soby Chacko; Chris Bono
//ifdef::backend-html5[]
//*{project-version}*

5
spring-pulsar-docs/src/main/asciidoc/preface.adoc
View File

@@ -1,3 +1,4 @@
This project provides a basic Spring friendly API for developing https://pulsar.apache.org/[Apache Pulsar] applications.
This project provides a basic Spring-friendly API for developing https://pulsar.apache.org/[Apache Pulsar] applications.
On a very high-level, Spring for Apache Pulsar provides a `PulsarTemplate` for publishing to a Pulsar topic and a `PulsarListener` annotation for consuming from it.
In addition, it also provides various convenient APIs for Sprind developers on ramp their development journey into Aapche Pulsar.
In addition, it also provides various convenient APIs for Spring developers to ramp up their development journey into Aapche Pulsar.

247
spring-pulsar-docs/src/main/asciidoc/pulsar.adoc
View File

@@ -1,5 +1,7 @@
[[pulsar]]
== Using Spring for Apache Pulsar
:javadocs: https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/api
:github: https://github.com/spring-projects-experimental/spring-pulsar
This section offers detailed explanations of the various concerns that impact using Spring for Apache Pulsar.
For a quick but less detailed introduction, see <<index.adoc#quick-intro>>.
@@ -7,102 +9,17 @@ For a quick but less detailed introduction, see <<index.adoc#quick-intro>>.
[[pulsar-client]]
=== Pulsar Client
When using `spring-pulsar-spring-boot-autoconfigure`, you get the `PulsarClient` auto-configured.
When using the Pulsar Spring Boot Starter, you get the `PulsarClient` auto-configured.
This is done through a factory bean called `PulsarClientFactoryBean`, which takes a configuration object `PulsarClientConfiguration`.
By default, the application tries to connect a local Pulsar instance available at `pulsar://localhost:6650`.
The properties listed below are available to be set on the Pulsar client.
The same defaults expected by the Pulsar client are honored when no configuration provided for these properties.
```
spring.pulsar.client.serviceUrl
spring.pulsar.client.authPluginClassName
spring.pulsar.client.authParams
spring.pulsar.client.operationTimeoutMs
spring.pulsar.client.statsIntervalSeconds
spring.pulsar.client.numIoThreads
spring.pulsar.client.useTcpNoDelayuseTls
spring.pulsar.client.tlsAllowInsecureConnection
spring.pulsar.client.tlsHostnameVerificationEnable
spring.pulsar.client.concurrentLookupRequest
spring.pulsar.client.maxLookupRequest
spring.pulsar.client.maxNumberOfRejectedRequestPerConnection
spring.pulsar.client.keepAliveIntervalSeconds
spring.pulsar.client.connectionTimeoutMs
spring.pulsar.client.requestTimeoutMs
spring.pulsar.client.initialBackoffIntervalNanos
spring.pulsar.client.maxBackoffIntervalNanos;
```
NOTE: By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`. However, there are many <<application-properties#appendix.application-properties.pulsar-client,application properties>> available to configure the client. When no configuration is provided for a property it's default value will be used.
[[pulsar-producer]]
=== Pulsar Producer
On the Pulsar producer side, Spring Boot auto-configuration will provide a `PulsarTemplate` which is backed by a `PulsarProducerFactory`.
We will see more details of these components later in this document, but in this section, let us look at the configuration properties available on the producer.
On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarOperations.html[several variants of 'send' methods] to publish records through its contract.
The same defaults expected by the Pulsar producer are honored when no configuration provided for these properties.
```
spring.pulsar.producer.topicName
spring.pulsar.producer.producerName
spring.pulsar.producer.sendTimeoutMs
spring.pulsar.producer.blockIfQueueFull
spring.pulsar.producer.maxPendingMessages
spring.pulsar.producer.maxPendingMessagesAcrossPartitions
spring.pulsar.producer.messageRoutingMode
spring.pulsar.producer.hashingScheme
spring.pulsar.producer.cryptoFailureAction
spring.pulsar.producer.batchingMaxPublishDelayMicros
spring.pulsar.producer.batchingMaxMessages
spring.pulsar.producer.batchingEnabled
spring.pulsar.producer.chunkingEnabled
spring.pulsar.producer.compressionType
spring.pulsar.producer.initialSubscriptionName
spring.pulsar.producer.accessMode
```
`PulsarTemplate` implements an interface called `PulsarOperations` and provides the following variants of the `send` method to publish records through its contract.
====
[source, java]
----
MessageId send(T message) throws PulsarClientException;
MessageId send(String topic, T message) throws PulsarClientException;
MessageId send(T message, MessageRouter messageRouter) throws PulsarClientException;
MessageId send(T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer) throws PulsarClientException;
MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException;
MessageId send(T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException;
MessageId send(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer) throws PulsarClientException;
MessageId send(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(String topic, T message) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(T message, MessageRouter messageRouter) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(String topic, T message, MessageRouter messageRouter) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException;
----
====
There are two categories of send API methods exposed by `PulsarTemplate` - `send` and `sendAsync`.
There are two categories of these send API methods - `send` and `sendAsync`.
The `send` methods are blocking calls using the synchronous sending capabilities on the Pulsar producer.
They return the `MessageId` of the message that was published once the message is persisted on the broker.
The `sendAsync` method calls are asynchronous calls that are non-blocking.
@@ -124,69 +41,29 @@ template.send(msg, (messageBuilder -> messageBuilder.key(myMessageKey)));
When using partitioned topics, then you can use custom partitioning routing when publishing records.
For this purpose, you can provide an implementation of a `MessageRouter` and pass it along with both `send` and `sendAsync` methods.
Note that, when using a `MessageRouter, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`.
Note that, when using a `MessageRouter`, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`.
NOTE: There are many <<application-properties#appendix.application-properties.pulsar-producer,application properties>> available to configure the producer.
[[pulsar-producer-factory]]
=== Pulsar Producer Factory
`PulsarTemplate` uses a `PulsarProducerFactory` for creating the underlying Pulsar producer.
When using Spring Boot through `spring-pulsar-spring-boot-autoconfigure`, then it automatically autoconfigures a `PulsarProducerFactory`.
Any producer properties mentioned above (using the prefix, `spring.pulsar.producer`) is passed along to the backing producer factory implementation along with a Pulsar client.
You can disable the autoconfigured producer factory by providing your own bean definition for `DefaultPulsarProducerFactory` in the application.
`PulsarProducerFactory` provides the following API methods to create a producer.
====
[source, java]
----
Producer<T> createProducer(String topic, Schema<T> schema) throws PulsarClientException;
Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter) throws PulsarClientException;
----
====
The `PulsarTemplate` relies on a `PulsarProducerFactory` for actually creating the underlying producer. Spring Boot auto-configuration also provides this producer factory. Additionally, you can configure the factory by specifying any of the available producer-centric <<application-properties#appendix.application-properties.pulsar-producer,application properties>>.
[[producer-caching]]
=== Pulsar Producer Caching
//TODO
**TODO**
[[pulsar-listener]]
=== Pulsar Listener
When it comes to Pulsar consumer, we recommend the end user applications to make use of the `PulsarListener` annotation.
In order to use `PulsarListener`, you need to use the `EnablePulsar` annotation.
In order to use `PulsarListener`, you need to use the `@EnablePulsar` annotation.
When using the Spring Boot support, it automatically enables this annotation and configures all the components necessary for `PulsarListener` such as the message listener infrastructure which is responsible for creating the Pulsar consumer.
`PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` in order to create and manage the Pulsar consumer.
This consumer factory is auto-configured through Spring Boot.
Following are the consumer properties that you can configure through the Boot support.
This consumer factory is also auto-configured through Spring Boot.
The same defaults expected by the Pulsar consumer are honored when no configuration provided for these properties.
```
spring.pulsar.consumer.topicNames
spring.pulsar.consumer.topicsPattern
spring.pulsar.consumer.subscriptionName
spring.pulsar.consumer.subscriptionType
spring.pulsar.consumer.receiverQueueSize
spring.pulsar.consumer.acknowledgementsGroupTimeMicros
spring.pulsar.consumer.negativeAckRedeliveryDelayMicros
spring.pulsar.consumer.maxTotalReceiverQueueSizeAcrossPartitions
spring.pulsar.consumer.consumerName
spring.pulsar.consumer.ackTimeoutMillis
spring.pulsar.consumer.tickDurationMillis
spring.pulsar.consumer.priorityLevel
spring.pulsar.consumer.cryptoFailureAction
spring.pulsar.consumer.properties
spring.pulsar.consumer.readCompacted
spring.pulsar.consumer.subscriptionInitialPosition
spring.pulsar.consumer.patternAutoDiscoveryPeriod
spring.pulsar.consumer.regexSubscriptionMode
spring.pulsar.consumer.autoUpdatePartitions
spring.pulsar.consumer.replicateSubscriptionState
spring.pulsar.consumer.autoAckOldestChunkedMessageOnQueueFull
spring.pulsar.consumer.maxPendingChunkedMessageexpireTimeOfIncompleteChunkedMessageMillis
spring.pulsar.consumer.maxPendingChunkedMessageexpireTimeOfIncompleteChunkedMessageMillis
```
NOTE: There are many consumer-centric <<application-properties#appendix.application-properties.pulsar-consumer,application properties>> available to configure these components.
Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section.
@@ -214,10 +91,12 @@ public void listen(String message) {
In this most basic form, you must provide the following two properties with their corresponding values.
```
spring.pulsar.consumer.topicNames=hello-pulsar
spring.pulsar.consumer.subscriptionName=hello-pulsar-subscription
```
[source,yaml,indent=0,subs="verbatim"]
----
spring.pulsar.consumer:
topic-names: hello-pulsar
subscription-name: hello-pulsar-subscription
----
In the `PulsarListener` method above, we receive the data as `String`, but we don't specify any schema types.
Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type.
@@ -261,9 +140,7 @@ Here is an example of using `PulsarListener` to consume records in batches.
@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true)
public void listen4(List<Foo> messages) {
System.out.println("records received :" + messages.size());
for (Foo message : messages) {
System.out.println("record : " + message);
}
messages.forEach((message) -> System.out.println("record : " + message));
}
----
====
@@ -282,9 +159,7 @@ The following also should work in which we use the `Messages` holder type provid
@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true)
public void listen4(Messages<Foo> messages) {
System.out.println("records received :" + messages.size());
for (Foo message : messages) {
System.out.println("record : " + message);
}
messages.forEach((message) -> System.out.println("record : " + message));
}
----
====
@@ -305,26 +180,13 @@ Once data is received, it is handed over to the selected message listener implem
The following message listener types are available when using Spring for Apache Pulsar.
====
[source, java]
----
interface PulsarRecordMessageListener<T> extends MessageListener<T>{
void received(Consumer<T> consumer, Message<T> msg);
}
* link:{github}/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java#L29[PulsarRecordMessageListener]
interface PulsarAcknowledgingMessageListener<T> extends PulsarRecordMessageListener<T> {
void received(Consumer<T> consumer, Message<T> msg, Acknowledgement acknowledgement);
}
* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarAcknowledgingMessageListener.java#L28[PulsarAcknowledgingMessageListener]
interface PulsarBatchMessageListener<T> extends PulsarRecordMessageListener<T> {
void received(Consumer<T> consumer, Messages<T> msg);
}
* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java#L36[PulsarBatchMessageListener]
interface PulsarBatchAcknowledgingMessageListener<T> extends PulsarBatchMessageListener<T> {
void received(Consumer<T> consumer, Messages<T> msg, Acknowledgement acknowledgement);
}
----
====
* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java#L28[PulsarBatchAcknowledgingMessageListener]
We will see the details about these various message listeners in the sections below.
@@ -362,9 +224,7 @@ Here is the `PulsarListener` example of consuming records in batches.
@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true)
public void listen4(List<Foo> messages) {
System.out.println("records received :" + messages.size());
for (Foo message : messages) {
System.out.println("record : " + message);
}
messages.forEach((message) -> System.out.println("record : " + message));
}
----
====
@@ -455,7 +315,9 @@ You can also negatively acknowledge with the `MessageId` for the batch listener.
In the sample below, we are publishing to a topic called `hello-pulsar-partitioned`.
It is a topic that is partitioned and for this sample we assume that the topic is already created with three partitions.
```
====
[source, java]
----
@SpringBootApplication
public class PulsarBootPartitioned {
@@ -505,7 +367,8 @@ 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.
@@ -524,7 +387,9 @@ We can switch to the `failover` subscription mode and add three separate consume
Here is an example.
```
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover")
public void listen1(String foo) {
System.out.println("Message Received 1: " + foo);
@@ -536,10 +401,11 @@ public void listen2(String foo) {
}
@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover")
public void liste3n(String foo) {
public void listen3(String foo) {
System.out.println("Message Received 3: " + foo);
}
```
----
====
When following this approach, you can see that a single partition always gets consumed by a dedicated consumer.
@@ -548,7 +414,9 @@ Keep in mind though, that when using the `shared` mode, you lose any ordering gu
Here is an example.
```
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "shared")
public void listen1(String foo) {
System.out.println("Message Received 1: " + foo);
@@ -558,46 +426,56 @@ public void listen1(String foo) {
public void listen2(String foo) {
System.out.println("Message Received 2: " + foo);
}
```
----
====
=== Accessing the Pulsar Message Object
In your `PulsarListener` method, you can receive the record directly as a Pulsar Message instead of the actual payload type.
Here is an example.
```
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public void listen(org.apache.pulsar.client.api.Message<String> message) {
System.out.println("Data Received: " + message.getValue());
}
```
----
====
=== Accessing the Pulsar Messages Object
When consuming messages in batch mode using `PulsarListener`, instead of receiving them as a `List, you can receive them as Pulsar Messages type.
Here is an example.
```
====
[source, java]
----
@PulsarListener(subscriptionName = "batch-subscription", topics = "hello-pulsar", batch = "true")
public void listen(org.apache.pulsar.client.api.Messages<String> messages) {
// Iterate on the messages
// Each iteration gives access to a org.apache.pulsar.client.api.Message object
}
```
----
====
=== Accessing the Pulsar Consumer Object
Sometimes, it is necessary to gain direct access to the Pulsar Consumer object.
Here is how you may do so.
```
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public void listen(String message, org.apache.pulsar.client.api.Consumer<String> consumer) {
System.out.println("Message Received: " + message);
ConsumerStats stats = consumer.getStats();
...
}
```
----
====
When accessing the `Consumer` object this way, make sure NOT to invoke any operations that would change the Consumer's cursor position by invoking any receive methods.
All such operations must be done by the container.
@@ -608,12 +486,15 @@ As indicated above, for normal Java types (the primitive ones), Spring Pulsar fr
However, for more complex types such as JSON or AVRO, you need to specify the schema type on the annotation.
Here is how you provide that.
```
====
[source, java]
----
@PulsarListener(subscriptionName = "json-subscription", topics = "hello-pulsar-json", schemaType = SchemaType.JSON)
public void listen(Foo foo) {
System.out.println("Message received: " + foo);
}
```
----
====
On the producer side also, for the Java primitive types, the framework can infer the Schema, but for any other types, you need set that on the `PulsarTemmplate`.

102
spring-pulsar-docs/src/main/asciidoc/quick-tour.adoc
View File

@@ -3,86 +3,68 @@
In this section, we will take a quick tour of Spring for Apache Pulsar.
#### Minimum Supported Versions
==== Minimum Supported Versions
The following are the minimum supported versions of the underlying libraries required by Spring for Apache Pulsar
The minimum supported versions for the underlying libraries required by the framework are as follows:
|===
| Library | Version
**JDK** - JDK 17
| Java
| 17
**Apache Pulsar** - 2.10.0
| Apache Pulsar
| 2.10.0
**Spring Boot** - 3.0.0
| Spring Boot
| 3.0.0
**Spring Framework** - 6.0.0
| Spring Framework
| 6.0.0
### Building the Project
| Gradle
| 7.x (7.5 or later)
|===
Spring for Apache Pulsar uses gradle as it's build tool.
Use the following command to do a full build of the project.
=== Building the Project
```
Spring for Apache Pulsar uses Gradle as it's build tool. Run the following command to do a full build of the project.
[indent=0]
----
./gradlew clean build
```
----
You can build without running tests by using the following command:
[indent=0]
----
./gradlew clean build -x test
----
The build will produce the following artifacts.
* spring-pulsar
* spring-pulsar-spring-boot-autoconfigure
### Maven Coordinates
Here are the maven coordinates of the modules, if you want to directly include them in an application.
Here are the maven coordinates of the artifacts needed for using this library.
```
<dependency>
<groupId>org.springframework.pulsar</groupId>
<artifactId>spring-pulsar</artifactId>
<version>0.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.pulsar</groupId>
<artifactId>spring-pulsar-boot-autoconfiguration</artifactId>
<version>0.1.0-SNAPSHOT</version>
</dependency>
```
If you include the `spring-pulsar-boot-autoconfiguration` module, then that brings the `spring-pulsar` module transitively.
We recommend the usage of a Spring-Boot-First-Approach for Spring for Apache Pulsar based application as that simplifies things tremendously.
Therefore, most often, simply including the `spring-pulsar-boot-autoconfiguration` module is all you need in the application's dependencies.
### Quick Sample
=== Quick Sample
In the following sample Spring Boot application, we show how to write a publisher and consumer using Spring for Apache Pulsar.
This is a complete application and does not require any additional configuration as long as you have Pulsar cluster running on the default location - `localhost:6650`.
```
NOTE: We recommend the usage of a Spring-Boot-First-Approach for Spring for Apache Pulsar based application as that simplifies things tremendously. To encourage this, a `spring-pulsar-spring-boot-starter` module is published that can easily be consumed by an application as a dependency.
[source,java,indent=0,pending-extract=true,subs="verbatim"]
----
@SpringBootApplication
public class PulsarBootHelloWorld {
public static void main(String[] args) {
SpringApplication.run(PulsarBootHelloWorld.class, args);
}
public static void main(String[] args) {
SpringApplication.run(PulsarBootHelloWorld.class, args);
}
@Bean
public ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
return args -> {
for (int i = 0; i < 10; i ++) {
pulsarTemplate.send("hello-pulsar", This is message " + (i + 1));
}
@Bean
ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
return (args) -> pulsarTemplate.send("hello-pulsar", "Hello Pulsar World!");
}
};
}
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public void listen(String message) {
System.out.println("Message Received: " + message);
}
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
void listen(String message) {
System.out.println("Message Received: " + message);
}
}
```
----
Let us go through the higher-level details of this application quickly.
Later on in this documentation, we will see these components in much more detail.

18
spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java
View File

@@ -33,7 +33,7 @@ import org.springframework.pulsar.core.PulsarTemplate;
@SpringBootApplication
public class SpringPulsarBootApp {
Logger logger = LoggerFactory.getLogger(SpringPulsarBootApp.class);
private final Logger logger = LoggerFactory.getLogger(SpringPulsarBootApp.class);
public static void main(String[] args) {
SpringApplication.run(SpringPulsarBootApp.class, args);
@@ -44,7 +44,7 @@ public class SpringPulsarBootApp {
* exclusive subscription to consume.
*/
@Bean
public ApplicationRunner runner1(PulsarTemplate<String> pulsarTemplate) {
ApplicationRunner runner1(PulsarTemplate<String> pulsarTemplate) {
String topic1 = "hello-pulsar-exclusive-1";
@@ -56,7 +56,7 @@ public class SpringPulsarBootApp {
}
@PulsarListener(subscriptionName = "subscription-1", topics = "hello-pulsar-exclusive-1")
public void listen1(String message) {
void listen1(String message) {
this.logger.info(message);
}
@@ -65,7 +65,7 @@ public class SpringPulsarBootApp {
* exclusive subscription to consume.
*/
@Bean
public ApplicationRunner runner2(PulsarTemplate<Integer> pulsarTemplate) {
ApplicationRunner runner2(PulsarTemplate<Integer> pulsarTemplate) {
String topic1 = "hello-pulsar-exclusive-2";
@@ -77,7 +77,7 @@ public class SpringPulsarBootApp {
}
@PulsarListener(subscriptionName = "subscription-2", topics = "hello-pulsar-exclusive-2")
public void listen2(Integer message) {
void listen2(Integer message) {
this.logger.info("Message received :" + message);
}
@@ -86,7 +86,7 @@ public class SpringPulsarBootApp {
* associated PulsarListener using an exclusive subscription.
*/
@Bean
public ApplicationRunner runner3(PulsarTemplate<Foo> pulsarTemplate) {
ApplicationRunner runner3(PulsarTemplate<Foo> pulsarTemplate) {
String topic = "hello-pulsar-exclusive-3";
return args -> {
@@ -99,7 +99,7 @@ public class SpringPulsarBootApp {
@PulsarListener(subscriptionName = "subscription-3", topics = "hello-pulsar-exclusive-3",
schemaType = SchemaType.JSON)
public void listen3(Foo message) {
void listen3(Foo message) {
this.logger.info("Message received :" + message);
}
@@ -107,7 +107,7 @@ public class SpringPulsarBootApp {
* Publish and then use PulsarListener in batch listening mode.
*/
@Bean
public ApplicationRunner runner4(PulsarTemplate<Foo> pulsarTemplate) {
ApplicationRunner runner4(PulsarTemplate<Foo> pulsarTemplate) {
String topic = "hello-pulsar-exclusive-4";
return args -> {
@@ -120,7 +120,7 @@ public class SpringPulsarBootApp {
@PulsarListener(subscriptionName = "subscription-4", topics = "hello-pulsar-exclusive-4",
schemaType = SchemaType.JSON, batch = true)
public void listen4(List<Foo> messages) {
void listen4(List<Foo> messages) {
this.logger.info("records received :" + messages.size());
for (Foo message : messages) {
this.logger.info("record : " + message);

10
spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java
View File

@@ -34,14 +34,14 @@ import org.springframework.pulsar.core.PulsarTemplate;
@SpringBootApplication
public class FailoverConsumerApp {
Logger logger = LoggerFactory.getLogger(FailoverConsumerApp.class);
private final Logger logger = LoggerFactory.getLogger(FailoverConsumerApp.class);
public static void main(String[] args) {
SpringApplication.run(FailoverConsumerApp.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition");
}
@Bean
public ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
String topic = "failover-demo-topic";
return args -> {
for (int i = 0; i < 10; i++) {
@@ -55,19 +55,19 @@ public class FailoverConsumerApp {
@PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic",
subscriptionType = "failover")
public void listen1(String foo) {
void listen1(String foo) {
this.logger.info("failover-listen1 : " + foo);
}
@PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic",
subscriptionType = "failover")
public void listen2(String foo) {
void listen2(String foo) {
this.logger.info("failover-listen2 : " + foo);
}
@PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic",
subscriptionType = "failover")
public void listen(String foo) {
void listen(String foo) {
this.logger.info("failover-listen3 : " + foo);
}