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

Add reactive docs (#250)

Browse Source 
Fixes #201
This commit is contained in:
Chris Bono
2022-12-12 00:32:14 -06:00
committed by GitHub
parent 7ee744bcfc
commit 0336228e99
17 changed files with 873 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,14 +1,14 @@
[appendix]
[[application-properties]]
[[appendix.application-properties]]
= Application Properties
include::attributes.adoc[]
You can specify various properties inside your `application.properties` file, inside your `application.yml` file, or as command line switches.
This appendix provides a list of Spring Pulsar properties and references to the underlying classes that consume them.
TIP: Spring Boot provides various conversion mechanisms with advanced value formatting. See <<features#features.external-config.typesafe-configuration-properties.conversion, the properties conversion section>> for more detail.
TIP: Spring Boot provides various conversion mechanisms with advanced value formatting.
See {spring-boot-docs}/#features.external-config.typesafe-configuration-properties.conversion[the properties conversion section] for more detail.
NOTE: Property contributions can come from additional jar files on your classpath, so this list is not exhaustive.
Also, you can define your own properties.
include::application-properties/pulsar-client.adoc[]
@@ -17,3 +17,9 @@ include::application-properties/pulsar-producer.adoc[]
include::application-properties/pulsar-consumer.adoc[]
include::application-properties/pulsar-administration.adoc[]
include::application-properties/pulsar-reactive-sender.adoc[]
include::application-properties/pulsar-reactive-consumer.adoc[]
include::application-properties/pulsar-reactive-reader.adoc[]

10
spring-pulsar-docs/src/main/asciidoc/attributes.adoc Normal file
View File

@@ -0,0 +1,10 @@
:toc: left
:toclevels: 6
:numbered:
:icons: font
:hide-uri-scheme:
:sectnums:
:sectnumlevels: 3
:github: https://github.com/spring-projects-experimental/spring-pulsar
:spring-boot-docs: https://docs.spring.io/spring-boot/docs/3.0.0/reference/htmlsingle

1
spring-pulsar-docs/src/main/asciidoc/authors.adoc Normal file
View File

@@ -0,0 +1 @@
Soby Chacko; Chris Bono; Alexander Preuß; Jay Bryant; Christophe Bornet

3
spring-pulsar-docs/src/main/asciidoc/copyright.adoc Normal file
View File

@@ -0,0 +1,3 @@
(C) 2022 VMware, Inc.
Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.

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

@@ -1,50 +1,41 @@
[[spring-pulsar-reference]]
= Spring for Apache Pulsar
:toc: left
:toclevels: 4
:numbered:
:icons: font
:hide-uri-scheme:
Soby Chacko; Chris Bono; Alexander Preuß; Jay Bryant; Christophe Bornet
//ifdef::backend-html5[]
//*{project-version}*
//
//NOTE: This documentation is also available as https://docs.spring.io/spring-pulsar/docs/{project-version}/reference/pdf/spring-pulsar-reference.pdf[PDF].
//endif::[]
//
//ifdef::backend-pdf[]
//NOTE: This documentation is also available as https://docs.spring.io/spring-pulsar/docs/{project-version}/reference/html/index.html[HTML].
//endif::[]
include::attributes.adoc[]
(C) 2022 VMware, Inc.
include::authors.adoc[]
Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.
include::copyright.adoc[]
== Preface
include::preface.adoc[]
[[quick-intro]]
== Introduction
This first part of the reference documentation is a high-level quick tour of Spring for Apache Pulsar.
include::quick-tour.adoc[]
include::intro.adoc[leveloffset=+1]
[[reference]]
== Reference
This part of the reference documentation goes through the details of the various components in Spring for Apache Pulsar.
include::pulsar.adoc[]
[[resources]]
include::reactive-pulsar.adoc[leveloffset=+2]
include::pulsar-admin.adoc[leveloffset=+2]
include::observability.adoc[leveloffset=+2]
:sectnums!:
[[other-resources]]
== Other Resources
In addition to this reference documentation, we recommend a number of other resources that may help you learn about Spring and Apache Pulsar.
- {github}[Spring for Apache Pulsar GitHub Repository]
- https://pulsar.apache.org/[Apache Pulsar Project Home Page]
- https://pulsar.apache.org/docs/client-libraries-java/[Apache Pulsar Java Client]
- https://github.com/spring-projects-experimental/spring-pulsar[Spring for Apache Pulsar GitHub Repository]
- https://github.com/apache/pulsar[Apache Pulsar GitHub Repository]
- https://github.com/apache/pulsar-client-reactive[Apache Pulsar Reactive Client GitHub Repository]
[[appendix]]
== Appendices
include::application-properties.adoc[leveloffset=+2]
include::non-ga-versions.adoc[leveloffset=+2]

41
spring-pulsar-docs/src/main/asciidoc/intro.adoc Normal file
View File

@@ -0,0 +1,41 @@
= Introduction
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 a Pulsar topic.
In addition, it also provides various convenience APIs for Spring developers to ramp up their development journey into Apache Pulsar.
== Minimum Supported Versions
The minimum supported versions for the underlying libraries required by the framework are as follows:
|===
| Library | Version
| Java
| 17
| Apache Pulsar
| 2.10.0
| Spring Boot
| 3.0.0
| Spring Framework
| 6.0.0
| Gradle
| 7.5
|===
== Building the Project
If you have cloned the project locally, follow these steps to build the project from the soure code.
Spring for Apache Pulsar uses Gradle as its 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
----

57
spring-pulsar-docs/src/main/asciidoc/non-ga-versions.adoc Normal file
View File

@@ -0,0 +1,57 @@
[appendix]
[[appendix.non-ga-versions]]
= Non-GA Versions
include::attributes.adoc[]
You can find snapshot or milestone versions of the dependencies in the following repositories:
[source,xml,indent=0,subs="verbatim,attributes",role="primary"]
.Maven
----
<repositories>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>apache-snapshots</id>
<name>Apache Snapshots</name>
<url>https://repository.apache.org/content/repositories/snapshots</url>
<releases>
<enabled>false</enabled>
</releases>
</repository>
</repositories>
----
[source,groovy,indent=0,subs="verbatim,attributes",role="secondary"]
.Gradle
----
repositories {
maven {
name = 'spring-milestones'
url = 'https://repo.spring.io/milestone'
}
maven {
name = 'spring-snapshots'
url = 'https://repo.spring.io/snapshot'
}
maven {
name = 'apache-snapshot'
url = 'https://repository.apache.org/content/repositories/snapshots'
}
}
----

96
spring-pulsar-docs/src/main/asciidoc/observability.adoc Normal file
View File

@@ -0,0 +1,96 @@
[[micrometer]]
= Observability
:github: https://github.com/spring-projects-experimental/spring-pulsar
Spring for Apache Pulsar includes a way to manage observability through https://micrometer.io/[Micrometer].
NOTE: Observability has not been added to the Reactive components yet
[[observation]]
== Micrometer Observations
The `PulsarTemplate` and `PulsarListener` are instrumented with the Micrometer observations API.
When a Micrometer `ObservationRegistry` bean is provided, send and receive operations are traced and timed.
=== Custom tags
The default implementation adds the `bean.name` tag for template observations and `listener.id` tag for listener observations.
To add other tags to timers and traces, configure a custom `PulsarTemplateObservationConvention` or `PulsarListenerObservationConvention` to the template or listener container, respectively.
TIP: You can subclass either `DefaultPulsarTemplateObservationConvention` or `DefaultPulsarListenerObservationConvention` or provide completely new implementations.
include::observation/_metrics.adoc[]
include::observation/_spans.adoc[]
See https://micrometer.io/docs/tracing[Micrometer Tracing] for more information.
=== Manual Configuration without Spring Boot
If you do not use Spring Boot, you need to configure and provide an `ObservationRegistry` as well as Micrometer Tracing. See https://micrometer.io/docs/tracing[Micrometer Tracing] for more information.
=== Auto-Configuration with Spring Boot
If you use Spring Boot, the Spring Boot Actuator auto-configures an instance of `ObservationRegistry` for you.
If `micrometer-core` is on the classpath, every stopped observation leads to a timer.
Spring Boot also auto-configures Micrometer Tracing for you. This includes support for Brave OpenTelemetry, Zipkin, and Wavefront. When using the Micrometer Observation API, finishing observations leads to spans reported to Zipkin or Wavefront. You can control tracing by setting properties under `management.tracing`. You can use Zipkin with `management.zipkin.tracing`, while Wavefront uses `management.wavefront`.
==== Example Configuration
The following example shows the steps to configure your Spring Boot application to use Zipkin with Brave.
. Add the required dependencies to your application (in Maven or Gradle, respectively):
+
[source,xml,indent=0,subs="verbatim,attributes",role="primary"]
.Maven
----
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing-bridge-brave</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter-brave</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-urlconnection</artifactId>
</dependency>
</dependencies>
----
+
[source,groovy,indent=0,subs="verbatim,attributes",role="secondary"]
.Gradle
----
dependencies {
implementation 'org.springframework.boot:spring-boot-starter-actuator'
implementation 'io.micrometer:micrometer-tracing-bridge-brave'
implementation 'io.zipkin.reporter2:zipkin-reporter-brave'
implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection'
}
----
+
NOTE
====
You need the `'io.zipkin.reporter2:zipkin-sender-urlconnection'` dependency only if your application does not have a configured WebClient or RestTemplate.
====
. Add the required properties to your application:
+
[source,yaml,indent=0,subs="verbatim"]
----
management:
tracing.enabled: true
zipkin:
tracing.endpoint: "http://localhost:9411/api/v2/spans"
----
+
The `tracing.endpoint` above expects Zipkin is running locally as described https://zipkin.io/pages/quickstart.html[here].
At this point, your application should record traces when you send and receive Pulsar messages. You should be able to view them in the Zipkin UI (at http://localhost:9411, when running locally).
TIP: You can also see the preceding configuration on the link:{github}/blob/main/spring-pulsar-sample-apps/README.adoc[Spring Pulsar Sample Apps].
The steps are very similar to configuring any of the other supported Tracing environments.

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

@@ -1,4 +0,0 @@
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 a Pulsar topic.
In addition, it also provides various convenience APIs for Spring developers to ramp up their development journey into Apache Pulsar.

43
spring-pulsar-docs/src/main/asciidoc/pulsar-admin.adoc Normal file
View File

@@ -0,0 +1,43 @@
[[pulsar-admin]]
= Pulsar Administration
:javadocs: https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/api
== Pulsar Admin Client
On the Pulsar administration side, Spring Boot auto-configuration provides a `PulsarAdministration` to manage Pulsar clusters.
The administration implements an interface called `PulsarAdminOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarAdminOperations.html[a `createOrModify` method] to handle topic administration through its contract.
When you use the Pulsar Spring Boot starter, you get the `PulsarAdministration` auto-configured.
By default, the application tries to connect to a local Pulsar instance at `http://localhost:8080`.
However, there are many application properties available to help you configure the client.
See the <<application-properties.adoc#appendix.application-properties.pulsar-administration,Appendix>> for application properties prefixed with `spring.pulsar.administration`.
[[pulsar-admin-authentication]]
=== Authentication
When accessing a Pulsar cluster that requires authentication, the admin client requires the same security configuration as the regular Pulsar client.
You can use the aforementioned <<pulsar.adoc#client-authentication,security configuration>> by replacing `spring.pulsar.client` with `spring.pulsar.administration`.
[[pulsar-auto-topic-creation]]
== Automatic Topic Creation
On initialization, the `PulsarAdministration` checks if there are any `PulsarTopic` beans in the application context.
For all such beans, the `PulsarAdministration` either creates the corresponding topic or, if necessary, modifies the number of partitions.
The following example shows how to add `PulsarTopic` beans to let the `PulsarAdministration` auto-create topics for you:
====
[source,java]
----
@Bean
PulsarTopic simpleTopic {
// This will create a non-partitioned topic in the public/default namespace
return PulsarTopic.builder("simple-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();
}
----
====

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

@@ -4,14 +4,14 @@
: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>>.
include::quick-tour.adoc[leveloffset=+3]
[[pulsar-client]]
==== Pulsar Client
When you use 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 called `PulsarClientConfiguration`. By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`. However, there are many application properties available to configure the client. See the <<application-properties.adoc#appendix.application-properties.pulsar-client,Appendix>> for more detail.
This is done through a factory bean called `PulsarClientFactoryBean`, which takes a configuration object called `PulsarClientConfiguration`. By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`. However, there are many application properties available to configure the client.
See the <<application-properties.adoc#appendix.application-properties.pulsar-client,Appendix>> for more detail.
[[client-authentication]]
===== Authentication
@@ -101,13 +101,15 @@ See the <<application-properties.adoc#appendix.application-properties.pulsar-pro
[[pulsar-producer-factory]]
==== Pulsar Producer Factory
The `PulsarTemplate` relies on a `PulsarProducerFactory` to actually create 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. See the <<application-properties.adoc#appendix.application-properties.pulsar-producer,Appendix>>.
The `PulsarTemplate` relies on a `PulsarProducerFactory` to actually create 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.
See the <<application-properties.adoc#appendix.application-properties.pulsar-producer,Appendix>>.
[[producer-caching]]
==== Pulsar Producer Caching
Each underlying Pulsar producer consumes resources. To improve performance and avoid continual creation of producers, the producer factory caches the producers that it creates. They are cached in an LRU fashion and evicted when they have not been used within a configured time period. The link:{github}/blob/8e33ac0b122bc0e75df299919c956cacabcc9809/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java#L159[cache key] is composed of just enough information to ensure that callers are returned the same producer on subsequent creation requests.
Additionally, you can configure the cache settings by specifying any of the `spring.producer.cache` prefixed application properties. See the <<application-properties.adoc#appendix.application-properties.pulsar-producer,Appendix>>.
Additionally, you can configure the cache settings by specifying any of the `spring.pulsar.producer.cache` prefixed application properties.
See the <<application-properties.adoc#appendix.application-properties.pulsar-producer,Appendix>>.
[[pulsar-listener]]
@@ -118,7 +120,8 @@ To use `PulsarListener`, you need to use the `@EnablePulsar` annotation.
When you use 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` to create and manage the Pulsar consumer.
This consumer factory is also auto-configured through Spring Boot. See the <<application-properties.adoc#appendix.application-properties.pulsar-consumer,Appendix>> for Pulsar consumer properties.
This consumer factory is also auto-configured through Spring Boot.
See the <<application-properties.adoc#appendix.application-properties.pulsar-consumer,Appendix>> for Pulsar consumer properties.
Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section:
@@ -287,7 +290,7 @@ void listen(String message) {
----
====
Note that the properties used are direct Pulsar consumer properties.
TIP: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` application configuration properties
[[pulsar-message-listener-container]]
==== Pulsar Message Listener Container
@@ -579,7 +582,8 @@ When calling `acknowledge()`, you need not receive the payload with the `Message
You can also call a different variant of `acknowledge` by providing the message ID: `acknowledge.acknowledge(message.getMessageId());`
When you use `acknowledge(messageId)`, you must receive the payload by using the `Message<?>` envelope.
Similar to what is possible for acknowledging, the `Acknowledgment` API also provides options for negatively acknowledging. See the nack methods shown earlier.
Similar to what is possible for acknowledging, the `Acknowledgment` API also provides options for negatively acknowledging.
See the nack methods shown earlier.
You can also call `acknowledge` directly on the Pulsar consumer:
@@ -1223,40 +1227,6 @@ ProducerInterceptor secondInterceptor() {
----
====
[[pulsar-admin]]
==== Pulsar Admin
On the Pulsar administration side, Spring Boot auto-configuration provides a `PulsarAdministration` to manage Pulsar clusters.
The administration implements an interface called `PulsarAdminOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarAdminOperations.html[a `createOrModify` method] to handle topic administration through its contract.
When you use the Pulsar Spring Boot starter, you get the `PulsarAdministration` auto-configured.
By default, the application tries to connect to a local Pulsar instance at `http://localhost:8080`. However, there are many application properties available to help you configure the client. See the <<application-properties.adoc#appendix.application-properties.pulsar-administration,Appendix>> for more detail.
===== Authentication
When accessing a Pulsar cluster that requires authentication, the admin client requires the same security configuration as the regular Pulsar client. You can use the aforementioned <<client-authentication,security configuration>> by replacing `spring.pulsar.client` with `spring.pulsar.administration`.
==== Automatic Topic creation
On initialization, the `PulsarAdministration` checks if there are any `PulsarTopic` beans in the application context.
For all such beans, the `PulsarAdministration` either creates the corresponding topic or, if necessary, modifies the number of partitions.
The following example shows how to add `PulsarTopic` beans to let the `PulsarAdministration` auto-create topics for you:
====
[source,java]
----
@Bean
PulsarTopic simpleTopic {
// This will create a non-partitioned topic in the public/default namespace
return PulsarTopic.builder("simple-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();
}
----
====
[[pulsar-headers]]
==== Pulsar Headers
@@ -1314,95 +1284,3 @@ When extracting the various headers, we do so as a `List<>` as well.
Spring Pulsar ensures that the headers list corresponds to the data list.
You can also extract headers in the same manner when you use the batch listener and receive payloads as `List<org.apache.pulsar.client.api.Message<?>`, `org.apache.pulsar.client.api.Messages<?>`, or `org.springframework.messaging.Messsge<?>`.
[[micrometer]]
=== Observability
Spring for Apache Pulsar includes a way to manage observability through https://micrometer.io/[Micrometer].
[[observation]]
==== Micrometer Observations
The `PulsarTemplate` and `PulsarListener` are instrumented with the Micrometer observations API.
When a Micrometer `ObservationRegistry` bean is provided, send and receive operations are traced and timed.
===== Custom tags
The default implementation adds the `bean.name` tag for template observations and `listener.id` tag for listener observations.
To add other tags to timers and traces, configure a custom `PulsarTemplateObservationConvention` or `PulsarListenerObservationConvention` to the template or listener container, respectively.
TIP: You can subclass either `DefaultPulsarTemplateObservationConvention` or `DefaultPulsarListenerObservationConvention` or provide completely new implementations.
include::observation/_metrics.adoc[leveloffset=+2]
include::observation/_spans.adoc[leveloffset=+2]
See https://micrometer.io/docs/tracing[Micrometer Tracing] for more information.
===== Manual Configuration without Spring Boot
If you do not use Spring Boot, you need to configure and provide an `ObservationRegistry` as well as Micrometer Tracing. See https://micrometer.io/docs/tracing[Micrometer Tracing] for more information.
===== Auto-Configuration with Spring Boot
If you use Spring Boot, the Spring Boot Actuator auto-configures an instance of `ObservationRegistry` for you.
If `micrometer-core` is on the classpath, every stopped observation leads to a timer.
Spring Boot also auto-configures Micrometer Tracing for you. This includes support for Brave OpenTelemetry, Zipkin, and Wavefront. When using the Micrometer Observation API, finishing observations leads to spans reported to Zipkin or Wavefront. You can control tracing by setting properties under `management.tracing`. You can use Zipkin with `management.zipkin.tracing`, while Wavefront uses `management.wavefront`.
====== Example Configuration
The following example shows the steps to configure your Spring Boot application to use Zipkin with Brave.
. Add the required dependencies to your application (in Maven or Gradle, respectively):
+
[source,xml,indent=0,subs="verbatim,attributes",role="primary"]
.Maven
----
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing-bridge-brave</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter-brave</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-urlconnection</artifactId>
</dependency>
</dependencies>
----
+
[source,groovy,indent=0,subs="verbatim,attributes",role="secondary"]
.Gradle
----
dependencies {
implementation 'org.springframework.boot:spring-boot-starter-actuator'
implementation 'io.micrometer:micrometer-tracing-bridge-brave'
implementation 'io.zipkin.reporter2:zipkin-reporter-brave'
implementation 'io.zipkin.reporter2:zipkin-sender-urlconnection'
}
----
+
NOTE
====
You need the `'io.zipkin.reporter2:zipkin-sender-urlconnection'` dependency only if your application does not have a configured WebClient or RestTemplate.
====
. Add the required properties to your application:
+
[source,yaml,indent=0,subs="verbatim"]
----
management:
tracing.enabled: true
zipkin:
tracing.endpoint: "http://localhost:9411/api/v2/spans"
----
+
The `tracing.endpoint` above expects Zipkin is running locally as described https://zipkin.io/pages/quickstart.html[here].
At this point, your application should record traces when you send and receive Pulsar messages. You should be able to view them in the Zipkin UI (at http://localhost:9411, when running locally).
TIP: You can also see the preceding configuration on the link:{github}/blob/main/spring-pulsar-sample-apps/README.adoc[Spring Pulsar Sample Apps].
The steps are very similar to configuring any of the other supported Tracing environments.

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

@@ -1,40 +1,14 @@
[[quick-tour]]
=== Quick Tour
= Quick Tour
:spring-pulsar-version: 0.1.0-SNAPSHOT
In this section, we take a quick tour of Spring for Apache Pulsar.
==== Minimum Supported Versions
The minimum supported versions for the underlying libraries required by the framework are as follows:
|===
| Library | Version
| Java
| 17
| Apache Pulsar
| 2.10.0
| Spring Boot
| 3.0.0
| Spring Framework
| 6.0.0
| Gradle
| 7.x (7.5 or later)
|===
=== Quick Sample
In the following sample Spring Boot application, we show how to write a publisher and a consumer that use Spring for Apache Pulsar.
We will take a quick tour of Spring for Apache Pulsar by showing a sample Spring Boot application that produces and consumes.
This is a complete application and does not require any additional configuration, as long as you have a Pulsar cluster running on the default location - `localhost:6650`.
NOTE: We recommend using a Spring-Boot-First approach for Spring for Apache Pulsar-based application, as that simplifies things tremendously. To do so, you can add the `spring-pulsar-spring-boot-starter` module as a dependency.
==== Dependencies
== Dependencies
Spring Boot applications need only the `spring-pulsar-spring-boot-starter` dependency. The following listings show how to define the dependency for Maven and Gradle, respectively:
@@ -58,44 +32,8 @@ dependencies {
}
----
===== Non-GA Versions
You can find snapshot or milestone versions of the dependency in the Spring Artifactory repository. The following listings show how to define the repositories for Maven and Gradle, respectively:
[source,xml,indent=0,subs="verbatim,attributes",role="primary"]
.Maven
----
<repositories>
...
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/snapshot</url>
<releases>
<enabled>false</enabled>
</releases>
</repository>
</repositories>
----
[source,groovy,indent=0,subs="verbatim,attributes",role="secondary"]
.Gradle
----
repositories {
...
maven { url 'https://repo.spring.io/milestone' }
maven { url 'https://repo.spring.io/snapshot' }
}
----
==== Application Code
== Application Code
The following listing shows the Spring Boot application case for the example:
@@ -110,10 +48,10 @@ public class PulsarBootHelloWorld {
@Bean
ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
return (args) -> pulsarTemplate.send("hello-pulsar", "Hello Pulsar World!");
return (args) -> pulsarTemplate.send("hello-pulsar-topic", "Hello Pulsar World!");
}
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
@PulsarListener(subscriptionName = "hello-pulsar-sub", topics = "hello-pulsar-topic")
void listen(String message) {
System.out.println("Message Received: " + message);
}
@@ -121,7 +59,7 @@ public class PulsarBootHelloWorld {
----
Let us quickly go through the higher-level details of this application.
<<index.adoc#reference,Later in this documentation>>, we see these components in much more detail.
Later in the documentation we see these components in much more detail.
In the preceding sample, we heavily rely on Spring Boot auto-configuration.
Spring Boot auto-configures several components for our application.
@@ -137,18 +75,3 @@ Behind the scenes, it creates a message listener container to create and manage
As with a regular Pulsar consumer, the default subscription type when using `PulsarListener` is the `Exclusive` mode.
As records are published in to the `hello-pulsar` topic, the `Pulsarlistener` consumes them and prints them on the console.
The framework also infers the schema type used from the data type that the `PulsarListner` method uses as the payload -- `String`, in this case.
=== Building the Project
If you have cloned the project locally, follow these steps to build the project from the soure code.
Spring for Apache Pulsar uses Gradle as its 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
----

504
spring-pulsar-docs/src/main/asciidoc/reactive-pulsar.adoc Normal file
View File

@@ -0,0 +1,504 @@
[[reactive-pulsar]]
= Reactive Support
:javadocs: https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/api
:github: https://github.com/spring-projects-experimental/spring-pulsar
The framework provides a Reactive counterpart for almost all supported features.
[TIP]
====
If you put the word `Reactive` in front of a provided imperative component, you will likely find its Reactive counterpart.
* `PulsarTemplate -> ReactivePulsarTemplate`
* `PulsarListener -> ReactivePulsarListener`
* `PulsarConsumerFactory -> ReactivePulsarConsumerFactory`
* etc..
====
However, the following is not yet supported:
* Error Handling in non-shared subscriptions
* Automatic ack/nacks in listeners
* Accessing Pulsar headers via `@Header` in streaming mode
* Observations
include::reactive-quick-tour.adoc[leveloffset=+1]
== Design
Here are a few key design points to keep in mind.
=== Apache Pulsar Reactive
The reactive support is ultimately provided by the https://github.com/apache/pulsar-client-reactive[Apache Pulsar Reactive client] whose current implementation is an adapter around the regular Pulsar client's asynchronous API.
This implies that the Reactive client requires the regular client.
NOTE: The current implementation is fully non-blocking but will likely change in the future to a fully native Reactive client
=== Additive Auto-Configuration
Due to the dependence on the regular (imperative) client, the Reactive auto-configuration provided by the framework is additive to the imperative auto-configuration.
In other words, The imperative starter only includes the imperative components but the reactive starter includes both imperative and reactive components.
[[reactive-pulsar-client]]
== Reactive Pulsar Client
When you use the Reactive Pulsar Spring Boot Starter, you get the `ReactivePulsarClient` auto-configured.
By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`.
However, there are many application properties (inherited from the adapted imperative client) available to configure.
See the <<application-properties.adoc#appendix.application-properties.pulsar-client,Appendix>> for properties prefixed with `spring.pulsar.client`.
[[reactive-client-authentication]]
=== Authentication
To connect to a Pulsar cluster that requires authentication, follow <<pulsar.adoc#client-authentication,the same steps>> as the imperative client.
Again, this is because the reactive client adapts the imperative client which handles all security configuration.
[[reactive-message-production]]
== Message Production
[[reactive-pulsar-template]]
=== ReactivePulsarTemplate
On the Pulsar producer side, Spring Boot auto-configuration provides a `ReactivePulsarTemplate` for publishing records. The template implements an interface called `ReactivePulsarOperations` and provides methods to publish records through its contract.
The template provides send methods that accept a single message and return a `Mono<MessageId>`.
It also provides send methods that accept multiple messages (in the form of the ReactiveStreams `Publisher` type) and return a `Flux<MessageId>`.
TIP: The send methods that do not have a topic input parameter require the topic name to be provided via the property `spring.pulsar.reactive.sender.topic-name`.
==== Fluent API
The template provides a {javadocs}/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests.
==== Message customization
You can specify a `MessageSpecBuilderCustomizer` to configure the outgoing message. For example, the following code shows how to send a keyed message:
====
[source, java]
----
template.newMessage(msg)
.withMessageCustomizer((mc) -> mc.key("foo-msg-key"))
.send();
----
====
==== Sender customization
You can specify a `ReactiveMessageSenderBuilderCustomizer` to configure the underlying Pulsar sender builder that ultimately constructs the sender used to send the outgoing message.
WARNING: Use with caution as this gives full access to the sender builder and invoking some of its methods (such as `create`) may have unintended side effects.
For example, the following code shows how to disable batching and enable chunking:
====
[source, java]
----
template.newMessage(msg)
.withSenderCustomizer((sc) -> sc.enableChunking(true).enableBatching(false))
.send();
----
====
This other example shows how to use custom routing when publishing records to partitioned topics.
Specify your custom `MessageRouter` implementation on the sender builder such as:
====
[source, java]
----
template.newMessage(msg)
.withSenderCustomizer((sc) -> sc.messageRouter(messageRouter))
.send();
----
====
TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.reactive.sender.message-routing-mode` is `custom`.
==== Schema
If you use Java primitive types, the framework auto-detects the schema for you, and you need not specify any schema types for publishing the data.
However, if you use any complex types (such as `JSON`, `AVRO`, `PROTOBUF`, and others), you need to set the proper schema type on the `ReactivePulsarTemplate` before invoking any send operations, as the following example shows for JSON:
====
[source, java]
----
template.setSchema(Schema.JSON(Foo.class));
----
====
IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PROTOBUF, and KEY_VALUE. For KEY_VALUE schemata, only INLINE encoding is supported.
[[reactive-sender-factory]]
=== ReactivePulsarSenderFactory
The `ReactivePulsarTemplate` relies on a `ReactivePulsarSenderFactory` to actually create the underlying sender.
Spring Boot provides this sender factory which can be configured with any of the <<application-properties.adoc#appendix.application-properties.pulsar-reactive-sender,`spring.pulsar.reactive.sender`>> prefixed application properties.
==== Producer Caching
Each underlying Pulsar producer consumes resources.
To improve performance and avoid continual creation of producers, the `ReactiveMessageSenderCache` in the underlying Apache Pulsar Reactive client caches the producers that it creates.
They are cached in an LRU fashion and evicted when they have not been used within a configured time period.
You can configure the cache settings by specifying any of the <<application-properties.adoc#appendix.application-properties.pulsar-reactive-sender,`spring.pulsar.reactive.sender.cache`>> prefixed application properties.
[[reactive-message-consumption]]
== Message Consumption
[[reactive-pulsar-listener]]
=== @ReactivePulsarListener
When it comes to Pulsar consumers, we recommend that end-user applications use the `ReactivePulsarListener` annotation.
To use `ReactivePulsarListener`, you need to use the `@EnableReactivePulsar` annotation.
When you use Spring Boot support, it automatically enables this annotation and configures all necessary components, such as the message listener infrastructure (which is responsible for creating the underlying Pulsar consumer).
Let us revisit the `ReactivePulsarListener` code snippet we saw in the quick-tour section:
====
[source, java]
----
@ReactivePulsarListener(subscriptionName = "hello-pulsar-sub", topics = "hello-pulsar-topic")
Mono<Void> listen(String message) {
System.out.println(message);
return Mono.empty();
}
----
====
NOTE: The listener method returns a `Mono<Void>` to signal whether the message was successfully processed. `Mono.empty()` indicates success (acknowledgment) and `Mono.error()` indicates failure (negative acknowledgment).
You can also further simplify this method:
====
[source, java]
----
@ReactivePulsarListener
Mono<Void> listen(String message) {
System.out.println(message);
return Mono.empty();
}
----
====
In this most basic form, you must still provide the topic name by setting the following property:
[source,yaml,indent=0,subs="verbatim"]
----
spring.pulsar.reactive.consumer:
topic-names: hello-pulsar-topic
----
NOTE: If `subscription-name` is not provided an auto-generated subscription name will be used.
In the `ReactivePulsarListener` method shown earlier, we receive the data as `String`, but we do not specify any schema types.
Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type.
The framework detects that you expect the `String` type and then infers the schema type based on that information.
Then it provides that schema to the consumer.
For all the primitive types in Java, the framework does this inference.
For any complex types (such as JSON, AVRO, and others), the framework cannot do this inference and the user needs to provide the schema type on the annotation using the `schemaType` property.
This example shows how we can consume complex types from a topic:
====
[source, java]
----
@ReactivePulsarListener(topics = "my-topic-2", schemaType = SchemaType.JSON)
Mono<Void> listen(Foo message) {
System.out.println(message);
return Mono.empty();
}
----
====
Note the addition of a `schemaType` property on `ReactivePulsarListener`.
That is because the library is not capable of inferring the schema type from the provided type: `Foo`. We must tell the framework what schema to use.
Let us look at a few more ways we can consume.
This example consumes the Pulsar message directly:
====
[source, java]
----
@ReactivePulsarListener(topics = "my-topic")
Mono<Void> listen(org.apache.pulsar.client.api.Message<String> message) {
System.out.println(message.getValue());
return Mono.empty();
}
----
====
This example consumes the record wrapped in a Spring messaging envelope:
====
[source, java]
----
@ReactivePulsarListener(topics = "my-topic")
Mono<Void> listen(org.springframework.messaging.Message<String> message) {
System.out.println(message.getPayload());
return Mono.empty();
}
----
====
==== Streaming
All of the above are examples of consuming a single record one-by-one.
However, one of the compelling reasons to use Reactive is for the streaming capability with backpressure support.
The following example uses `ReactivePulsarListener` to consume a stream of POJOs:
====
[source, java]
----
@ReactivePulsarListener(topics = "streaming-1", stream = true)
Flux<MessageResult<Void>> listen(Flux<Message<String>> messages) {
return messages
.doOnNext((msg) -> System.out.println("Received: " + msg.getValue()))
.map(MessageResult::acknowledge);
----
====
Here we receive the records as a `Flux` of messages.
In addition, to enable stream consumption at the `ReactivePulsarListener` level, you need to set the `stream` property on the annotation to `true`.
NOTE: The listener method returns a `Flux<MessageResult<Void>>` where each element represents a processed message and holds the message id, value and whether it was acknowledged. The `MessageResult` has a set of static factory methods that can be used to create the appropriate `MessageResult` instance.
Based on the actual type of the messages in the `Flux`, the framework tries to infer the schema to use.
If it contains a complex type, you still need to provide the `schemaType` on `ReactivePulsarListener`.
The following listener uses the Spring messaging `Message` envelope with a complex type :
====
[source, java]
----
@ReactivePulsarListener(topics = "streaming-2", stream = true, schemaType = SchemaType.JSON)
Flux<MessageResult<Void>> listen2(Flux<org.springframework.messaging.Message<Foo>> messages) {
return messages
.doOnNext((msg) -> System.out.println("Received: " + msg.getPayload()))
.map(MessageResult::acknowledge);
}
----
====
==== Configuration - Application Properties
The listener ultimately relies on `ReactivePulsarConsumerFactory` to create and manage the underlying Pulsar consumer.
Spring Boot provides this consumer factory which can be configured with any of the <<application-properties.adoc#appendix.application-properties.pulsar-reactive-consumer,`spring.pulsar.reactive.consumer`>> prefixed application properties.
[[reactive-consumer-customizer]]
==== Consumer Customization
You can specify a `ReactiveMessageConsumerBuilderCustomizer` to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages.
WARNING: Use with caution as this gives full access to the consumer builder and invoking some of its methods (such as `create`) may have unintended side effects.
For example, the following code shows how to set the initial position of the subscription to the earliest messaage on the topic.
====
[source, java]
----
@ReactivePulsarListener(topics = "hello-pulsar-topic", consumerCustomizer = "myConsumerCustomizer")
Mono<Void> listen(String message) {
System.out.println(message);
return Mono.empty();
}
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> myConsumerCustomizer() {
return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
}
----
====
You can also use the customizer to provide direct Pulsar consumer properties to the consumer builder.
This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `ReactivePulsarListener` methods whose configuration varies.
The following customizer example uses direct Pulsar consumer properties:
====
[source, java]
----
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> directConsumerPropsCustomizer() {
return b -> b.property("subscriptionName", "subscription-1").property("topicNames", "foo-1");
}
----
====
CAUTION: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.reactive.consumer` Spring Boot configuration properties
[[reactive-message-listener-container]]
=== Message Listener Container Infrastructure
In most scenarios, we recommend using the `ReactivePulsarListener` annotation directly for consuming from a Pulsar topic as that model covers a broad set of application use cases.
However, it is important to understand how `ReactivePulsarListener` works internally.
The message listener container is at the heart of message consumption when you use Spring for Apache Pulsar.
The `ReactivePulsarListener` uses the message listener container infrastructure behind the scenes to create and manage the underlying Pulsar consumer.
==== ReactivePulsarMessageListenerContainer
The contract for this message listener container is provided through `ReactivePulsarMessageListenerContainer` whose default implementation creates a reactive Pulsar consumer and wires up a reactive message pipeline that uses the created consumer.
==== ReactiveMessagePipeline
The pipeline is a feature of the underlying Apache Pulsar Reactive client which does the heavy lifting of receiving the data in a reactive manner and then handing it over to the provided message handler. The reactive message listener container implementation is much simpler because the pipeline handles the majority of the work.
==== ReactivePulsarMessageHandler
The "listener" aspect is provided by the `ReactivePulsarMessageHandler` of which there are two provided implementations:
* `ReactivePulsarOneByOneMessageHandler` - handles a single message one-by-one
* `ReactivePulsarStreamingHandler` - handles multiple messages via a `Flux`
[[reactive-concurrency]]
=== Concurrency
When consuming records in streaming mode (`stream = true`) concurrency comes naturally via the underlying Reactive support in the client implementation.
However, when handling messages one-by-one, the concurrency can be specified to increase processing throughput.
Simply set the `concurrency` property on `@ReactivePulsarListener`.
Additionally, when `concurrency > 1` you can ensure messages are ordered by key on each parallel by setting `useKeyOrderedProcessing = "true"` on the annotation.
Again, the `ReactiveMessagePipeline` does the heavy lifting, we simply set the properties on it.
====
**QUESTION**
How does the subscription type affect the concurrency setting?
How does the number of partitions and subscription type interact w/ the concurrency setting in the ReactiveMessagePipeline?
The imperative counterpart says these things: https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/reference/html/#_concurrentpulsarmessagelistenercontainer
====
[[reactive-pulsar-headers]]
=== Pulsar Headers
The Pulsar message metadata can be consumed as Spring message headers.
The list of available headers can be found in https://github.com/spring-projects-experimental/spring-pulsar/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java[PulsarHeaders.java].
==== Accessing In OneByOne Listener
The following example shows how you can access Pulsar Headers when using a one-by-one message listener:
====
[source,java]
----
@ReactivePulsarListener(topics = "some-topic")
Mono<Void> listen(String data,
@Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,
@Header("foo") String foo) {
System.out.println("Received " + data + " w/ id=" + messageId + " w/ foo=" + foo);
return Mono.empty();
}
----
====
In the preceding example, we access the values for the `messageId` message metadata as well as a custom message property named `foo`.
The Spring `@Header` annotation is used for each header field.
You can also use Pulsar's `Message` as the envelope to carry the payload.
When doing so, the user can directly call the corresponding methods on the Pulsar message for retrieving the metadata.
However, as a convenience, you can also retrieve it by using the `Header` annotation.
Note that you can also use the Spring messaging `Message` envelope to carry the payload and then retrieve the Pulsar headers by using `@Header`.
==== Accessing In Streaming Listener
When using a streaming message listener the header support is limited.
Only when the `Flux` contains Spring `org.springframework.messaging.Message` elements will the headers be populated.
Additionally, the Spring `@Header` annotation can not be used to retrieve the data.
You must directly call the corresponding methods on the Spring message to retrieve the data.
[[reactive-message-ack]]
=== Message Acknowledgment
Unlike its imperative counterpart, message acknowledgment must be handled manually (albeit indirectly) when consuming in a Reactive fashion.
The listener method must return a signal of success or failure.
The container implementation then uses that signal to perform the ack or nack operation.
==== OneByOne Listener
The single message (aka OneByOne) message listener method returns a `Mono<Void>` to signal whether the message was successfully processed. `Mono.empty()` indicates success (acknowledgment) and `Mono.error()` indicates failure (negative acknowledgment).
==== Streaming Listener
The streaming listener method returns a `Flux<MessageResult<Void>>` where each `MessageResult` element represents a processed message and holds the message id, value and whether it was acknowledged. The `MessageResult` has a set of `acknowledge` and `negativeAcknowledge` static factory methods that can be used to create the appropriate `MessageResult` instance
[[reactive-redelivery]]
=== Message Redelivery and Error Handling
Apache Pulsar provides various native strategies for message redelivery and error handling.
We will take a look at them and see how to use them through Spring for Apache Pulsar.
==== Acknowledgment Timeout
By default, Pulsar consumers do not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer.
When you use Spring for Apache Pulsar, you can enable this property by setting the `spring.pulsar.reactive.consumer.ack-timeout` Boot property.
If this property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered.
You can also specify this property directly as a Pulsar consumer property via a <<reactive-consumer-customizer,consumer customizer>> such as:
====
[source, java]
----
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
return b -> b.property("ackTimeout", "60s");
}
----
====
==== Negative Acknowledgment Redelivery Delay
When acknowledging negatively, Pulsar consumer lets you specify how the application wants the message to be re-delivered.
The default is to redeliver the message in one minute, but you can change it by setting `spring.pulsar.reactive.consumer.negative-ack-redelivery-delay`.
You can also set it directly as a Pulsar consumer property via a <<reactive-consumer-customizer,consumer customizer>> such as:
====
[source, java]
----
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
return b -> b.property("negativeAckRedeliveryDelay", "10ms");
}
----
====
==== Dead Letter Topic
Apache Pulsar lets applications use a dead letter topic on consumers with a `Shared` subscription type.
For the `Exclusive` and `Failover` subscription types, this feature is not available.
The basic idea is that, if a message is retried a certain number of times (maybe due to an ack timeout or nack redelivery), once the number of retries are exhausted, the message can be sent to a special topic called the dead letter queue (DLQ).
Let us see some details around this feature in action by inspecting some code snippets:
====
[source, java]
----
@Configuration(proxyBeanMethods = false)
class DeadLetterPolicyConfig {
@ReactivePulsarListener(
topics = "topic-with-dlp",
subscriptionType = SubscriptionType.Shared,
deadLetterPolicy = "myDeadLetterPolicy",
consumerCustomizer = "ackTimeoutCustomizer" )
void listen(String msg) {
throw new RuntimeException("fail " + msg);
}
@ReactivePulsarListener(topics = "my-dlq-topic")
void listenDlq(String msg) {
System.out.println("From DLQ: " + msg);
}
@Bean
DeadLetterPolicy myDeadLetterPolicy() {
return DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic("my-dlq-topic").build();
}
@Bean
ReactiveMessageConsumerBuilderCustomizer<String> ackTimeoutCustomizer() {
return b -> b.property("ackTimeout", "1s");
}
}
----
====
First, we have a special bean for `DeadLetterPolicy`, and it is named as `deadLetterPolicy` (it can be any name as you wish).
This bean specifies a number of things, such as the max delivery (10, in this case) and the name of the dead letter topic -- `my-dlq-topic`, in this case.
If you do not specify a DLQ topic name, it defaults to `<topicname>-<subscriptionname>-DLQ` in Pulsar.
Next, we provide this bean name to `ReactivePulsarListener` by setting the `deadLetterPolicy` property.
Note that the `ReactivePulsarListener` has a subscription type of `Shared`, as the DLQ feature only works with shared subscriptions.
This code is primarily for demonstration purposes, so we provide an `ackTimeout` value of 1 second.
The idea is that the code throws the exception and, if Pulsar does not receive an ack within 1 second, it does a retry.
If that cycle continues ten times (as that is our max redelivery count in the `DeadLetterPolicy`), the Pulsar consumer publishes the messages to the DQL topic.
We have another `ReactivePulsarListener` that listens on the DLQ topic to receive data as it is published to the DLQ topic.
.Special note on DLQ topics when using partitioned topics
****
If the main topic is partitioned, behind the scenes, each partition is treated as a separate topic by Pulsar.
Pulsar appends `partition-<n>`, where `n` stands for the partition number to the main topic name.
The problem is that, if you do not specify a DLQ topic (as opposed to what we did above), Pulsar publishes to a default topic name that has this ``partition-<n>` info in it -- for example: `topic-with-dlp-partition-0-deadLetterPolicySubscription-DLQ`.
The easy way to solve this is to provide a DLQ topic name always.
****
[[reactive-pulsar-reader]]
=== Pulsar Reader Support
The framework provides support for using https://pulsar.apache.org/docs/2.10.x/concepts-clients/#reader-interface[Pulsar Reader] in a Reactive fashion via the `ReactivePulsarReaderFactory`.
Spring Boot provides this reader factory which can be configured with any of the <<application-properties.adoc#appendix.application-properties.pulsar-reactive-reader,`spring.pulsar.reactive.reader`>> prefixed application properties.

65
spring-pulsar-docs/src/main/asciidoc/reactive-quick-tour.adoc Normal file
View File

@@ -0,0 +1,65 @@
[[quick-tour-reactive]]
= Quick Tour
:spring-pulsar-version: 0.1.0-SNAPSHOT
We will take a quick tour of the Reactive support in Spring for Apache Pulsar by showing a sample Spring Boot application that produces and consumes in a Reactive fashion.
This is a complete application and does not require any additional configuration, as long as you have a Pulsar cluster running on the default location - `localhost:6650`.
NOTE: We recommend using a Spring-Boot-First approach for Spring for Apache Pulsar-based applications, as that simplifies things tremendously. To do so, you can add the `spring-pulsar-reactive-spring-boot-starter` module as a dependency.
== Dependencies
Spring Boot applications need only the `spring-pulsar-reactive-spring-boot-starter` dependency. The following listings show how to define the dependency for Maven and Gradle, respectively:
[source,xml,indent=0,subs="verbatim,attributes",role="primary"]
.Maven
----
<dependencies>
<dependency>
<groupId>org.springframework.pulsar</groupId>
<artifactId>spring-pulsar-reactive-spring-boot-starter</artifactId>
<version>{spring-pulsar-version}</version>
</dependency>
</dependencies>
----
[source,groovy,indent=0,subs="verbatim,attributes",role="secondary"]
.Gradle
----
dependencies {
implementation 'org.springframework.pulsar:spring-pulsar-reactive-spring-boot-starter:{spring-pulsar-version}'
}
----
== Application Code
Here is the application source code:
[source,java,indent=0,pending-extract=true,subs="verbatim"]
----
@SpringBootApplication
public class ReactiveSpringPulsarHelloWorld {
public static void main(String[] args) {
SpringApplication.run(ReactiveSpringPulsarHelloWorld.class, args);
}
@Bean
ApplicationRunner runner(ReactivePulsarTemplate<String> pulsarTemplate) {
return (args) -> pulsarTemplate.send("hello-pulsar-topic", "Hello Reactive Pulsar World!").subscribe();
}
@ReactivePulsarListener(subscriptionName = "hello-pulsar-sub", topics = "hello-pulsar-topic")
Mono<Void> listen(String message) {
System.out.println("Reactive listener received: " + message);
return Mono.empty();
}
}
----
That is it, with just a few lines of code we have a working Spring Boot app that is producing and consuming messages from a Pulsar topic in a Reactive fashion.
Once started, the application uses a `ReactivePulsarTemplate` to send messages to the `hello-pulsar-topic`.
It then consumes from the `hello-pulsar-topic` using a `@ReactivePulsarListener`.
NOTE: One of the key ingredients to the simplicity is the Spring Boot starter which auto-configures and provides the required components to the application

2
spring-pulsar-sample-apps/sample-app1/build.gradle
View File

@@ -6,7 +6,7 @@ plugins {
description = 'Spring Pulsar Sample Application (Send and Receive)'
dependencies {
api project(':spring-pulsar-spring-boot-starter')
implementation project(':spring-pulsar-spring-boot-starter')
implementation 'com.google.code.findbugs:jsr305'
// observability

2
spring-pulsar-sample-apps/sample-app2/build.gradle
View File

@@ -6,7 +6,7 @@ plugins {
description = 'Spring Pulsar Sample Applications (Custom Routing)'
dependencies {
api project(':spring-pulsar-spring-boot-starter')
implementation project(':spring-pulsar-spring-boot-starter')
implementation 'com.google.code.findbugs:jsr305'
// observability

2
spring-pulsar-sample-apps/sample-reactive/build.gradle
View File

@@ -6,7 +6,7 @@ plugins {
description = 'Reactive Spring Pulsar Sample Application'
dependencies {
api project(':spring-pulsar-reactive-spring-boot-starter')
implementation project(':spring-pulsar-reactive-spring-boot-starter')
implementation 'com.google.code.findbugs:jsr305'
}