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

Add reference guide and javadocs

Browse Source
This commit is contained in:
Soby Chacko
2022-07-26 19:37:25 -04:00
committed by Chris Bono
parent 931f8f8e2c
commit 7e0c81ad1c
9 changed files with 1058 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
build.gradle
View File

@@ -344,6 +344,17 @@ project ('spring-pulsar-boot-autoconfigure') {
}
}
project ('spring-pulsar-docs') {
description = 'Spring Pulsar Code Snippets For Docs'
dependencies {
api "org.springframework.boot:spring-boot-starter:$springBootVersion"
api project (':spring-pulsar')
optionalApi 'com.fasterxml.jackson.core:jackson-core'
optionalApi 'com.fasterxml.jackson.core:jackson-databind'
}
}
project ('spring-pulsar-sample-apps') {
description = 'Spring Pulsar Sample Applications'
@@ -380,3 +391,198 @@ if (project.hasProperty('artifactoryUsername')) {
}
}
}
configurations {
docs
}
dependencies {
docs "io.spring.docresources:spring-doc-resources:${docResourcesVersion}@zip"
}
task prepareAsciidocBuild(type: Sync) {
dependsOn configurations.docs
from {
configurations.docs.collect { zipTree(it) }
}
duplicatesStrategy = DuplicatesStrategy.EXCLUDE
from 'spring-pulsar-docs/src/main/asciidoc/'
into "$buildDir/asciidoc"
}
asciidoctorPdf {
dependsOn prepareAsciidocBuild
baseDirFollowsSourceFile()
configurations 'asciidoctorExt'
asciidoctorj {
sourceDir "$buildDir/asciidoc"
inputs.dir(sourceDir)
sources {
include 'index.adoc'
}
options doctype: 'book'
attributes 'icons': 'font',
'sectanchors': '',
'sectnums': '',
'toc': '',
'source-highlighter' : 'coderay',
revnumber: project.version,
'project-version': project.version
}
}
asciidoctorj {
version = '2.4.1'
// fatalWarnings ".*"
options doctype: 'book', eruby: 'erubis'
attributes([
icons: 'font',
idprefix: '',
idseparator: '-',
docinfo: 'shared',
revnumber: project.version,
sectanchors: '',
sectnums: '',
'source-highlighter': 'highlight.js',
highlightjsdir: 'js/highlight',
'highlightjs-theme': 'googlecode',
stylesdir: 'css/',
stylesheet: 'stylesheet.css',
'spring-version': project.version,
'project-version': project.version
])
}
asciidoctor {
dependsOn asciidoctorPdf
baseDirFollowsSourceFile()
configurations 'asciidoctorExt'
sourceDir = file("$buildDir/asciidoc")
sources {
include '*.adoc'
}
resources {
from(sourceDir) {
include 'images/*', 'css/**', 'js/**'
}
}
}
task reference(dependsOn: asciidoctor) {
group = 'Documentation'
description = 'Generate the reference documentation'
}
reference.onlyIf { "$System.env.NO_REFERENCE_TASK" != 'true' || project.hasProperty('ignoreEnvToStopReference') }
task api(type: Javadoc) {
group = 'Documentation'
description = 'Generates aggregated Javadoc API documentation.'
title = "${rootProject.description} ${version} API"
options {
encoding = 'UTF-8'
memberLevel = org.gradle.external.javadoc.JavadocMemberLevel.PROTECTED
author = true
header = rootProject.description
use = true
overview = 'src/api/overview.html'
splitIndex = true
links(project.ext.javadocLinks)
addBooleanOption('Xdoclint:syntax', true) // only check syntax with doclint
}
source subprojects.collect { project ->
project.sourceSets.main.allJava
}
classpath = files(subprojects.collect { project ->
project.sourceSets.main.compileClasspath
})
destinationDir = new File(buildDir, 'api')
}
task docsZip(type: Zip, dependsOn: [reference]) {
group = 'Distribution'
archiveClassifier = 'docs'
description = "Builds -${archiveClassifier} archive containing api and reference " +
"for deployment at static.spring.io/spring-pulsar/docs."
from('src/dist') {
include 'changelog.txt'
}
from(api) {
into 'api'
}
from ('build/docs/asciidoc') {
into 'reference/html'
}
from ('build/docs/asciidocPdf') {
include 'index.pdf'
rename 'index.pdf', 'spring-pulsar-reference.pdf'
into 'reference/pdf'
}
}
task distZip(type: Zip, dependsOn: [docsZip]) { //, schemaZip]) {
group = 'Distribution'
archiveClassifier = 'dist'
description = "Builds -${archiveClassifier} archive, containing all jars and docs, " +
"suitable for community download page."
ext.baseDir = "${project.name}-${project.version}"
from('src/dist') {
include 'readme.txt'
include 'notice.txt'
into "${baseDir}"
}
from("$project.rootDir") {
include 'LICENSE.txt'
into "${baseDir}"
}
from(zipTree(docsZip.archivePath)) {
into "${baseDir}/docs"
}
subprojects.each { subproject ->
into ("${baseDir}/libs") {
from subproject.jar
from subproject.sourcesJar
from subproject.javadocJar
}
}
}
artifacts {
archives distZip
archives docsZip
}
task dist(dependsOn: assemble) {
group = 'Distribution'
description = 'Builds -dist, -docs distribution archives.'
}
apply from: "${rootProject.projectDir}/publish-maven.gradle"
publishing {
publications {
mavenJava(MavenPublication) {
artifact docsZip
artifact distZip
}
}
}
apply from: "${rootDir}/gradle/docs.gradle"

49
publish-maven.gradle Normal file
View File

@@ -0,0 +1,49 @@
apply plugin: 'maven-publish'
apply plugin: 'com.jfrog.artifactory'
publishing {
publications {
mavenJava(MavenPublication) {
pom {
afterEvaluate {
name = project.description
description = project.description
}
url = linkScmUrl
organization {
name = 'Spring IO'
url = 'https://spring.io/projects/spring-pulsar'
}
licenses {
license {
name = 'Apache License, Version 2.0'
url = 'https://www.apache.org/licenses/LICENSE-2.0.txt'
distribution = 'repo'
}
}
scm {
url = linkScmUrl
connection = linkScmConnection
developerConnection = linkScmDevConnection
}
issueManagement {
system = 'GitHub'
url = linkIssue
}
}
versionMapping {
usage('java-api') {
fromResolutionResult()
}
usage('java-runtime') {
fromResolutionResult()
}
}
}
}
}
artifactoryPublish {
dependsOn build
publications(publishing.publications.mavenJava)
}

1
settings.gradle
View File

@@ -21,3 +21,4 @@ rootProject.name = 'spring-pulsar-dist'
include 'spring-pulsar'
include 'spring-pulsar-boot-autoconfigure'
include 'spring-pulsar-sample-apps'
include 'spring-pulsar-docs'

21
spring-pulsar-docs/build.gradle Normal file
View File

@@ -0,0 +1,21 @@
jar {
enabled = false
}
sourcesJar {
enabled = false
}
javadocJar {
enabled = false
}
publishMavenJavaPublicationToMavenLocal {
enabled = false
}
project.afterEvaluate {
project.tasks.artifactoryPublish.enabled(false)
}

31
spring-pulsar-docs/src/main/asciidoc/css/stylesheet.css Normal file
View File

@@ -0,0 +1,31 @@
@import 'css/spring.css';
.listingblock .switch {
border-style: none;
display: inline-block;
position: relative;
bottom: -3px;
}
.listingblock .switch--item {
padding: 10px;
background-color: #e6e1dc;
color: #282c34;
display: inline-block;
cursor: pointer;
border-top-left-radius: 4px;
border-top-right-radius: 4px;
}
.listingblock .switch--item:not(:first-child) {
border-style: none;
}
.listingblock .switch--item.selected {
background-color: #282c34;
color: #e6e1dc;
}
.listingblock pre.highlightjs {
padding: 0;
}

50
spring-pulsar-docs/src/main/asciidoc/index.adoc Normal file
View File

@@ -0,0 +1,50 @@
[[spring-pulsar-reference]]
= Spring for Apache Pulsar
:toc: left
:toclevels: 4
:numbered:
:icons: font
:hide-uri-scheme:
Soby Chacko
//ifdef::backend-html5[]
//*{project-version}*
//
//NOTE: This documentation is also available as https://docs.spring.io/spring-kafka/docs/{project-version}/reference/pdf/spring-kafka-reference.pdf[PDF].
//endif::[]
//
//ifdef::backend-pdf[]
//NOTE: This documentation is also available as https://docs.spring.io/spring-kafka/docs/{project-version}/reference/html/index.html[HTML].
//endif::[]
(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.
== Preface
include::preface.adoc[]
== Introduction
This first part of the reference documentation is a high-level quick-tour of Spring for Apache Pulsar.
include::quick-tour.adoc[]
== Reference
This part of the reference documentation goes through the details of the various components in Spring for Apache Pulsar.
include::pulsar.adoc[]
[[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.
- 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]

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

@@ -0,0 +1,3 @@
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.

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

@@ -0,0 +1,592 @@
[[pulsar]]
== Using Spring for Apache 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 <<quick-tour>>.
[[pulsar-client]]
=== Pulsar Client
When using `spring-pulsar-boot-autoconfigure`, 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;
```
[[pulsar-producer]]
=== Pulsar Producer
On the Pulsar producer side, Sprig 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.
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;
send(String topic, T message, 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);
CompletableFuture<MessageId> sendAsync(String topic, T message, MessageRouter messageRouter);
----
====
There are two categories of send API methods exposed by `PulsarTemplate` - `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.
They return a `CompletableFuture` using which you can asynchronously receive the message id once the messages are published.
Both `send` and `sendAsync` methods have a variant that allows to publish simply with the message.
When you do that, the application must provide the topic name using the property `spring.pulsar.producer.topicName`.
There is another variant for both flavors that takes the topic name as an argument.
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`.
[[pulsar-producer-factory]]
=== Pulsar Producer Factory
`PulsarTemplate` uses a `PulsarProducerFactory` for creating the underlying Pulsar producer.
When using Spring Boot through `spring-pulsar-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;
----
====
[[producer-caching]]
=== Pulsar Producer Caching
//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.
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.
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
```
Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section.
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public void listen(String message) {
System.out.println("Message Received: " + message);
}
----
====
This can even be further simplified as below.
====
[source, java]
----
@PulsarListener
public void listen(String message) {
System.out.println("Message Received: " + 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
```
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.
The framework detects that you are expecting 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 etc. the framework cannot do this inference and the user needs to provide the schema type on the annotation using the `schemaType` property.
Here is another `PulsarListener` method, that takes an `Integer`.
====
[source, java]
----
@PulsarListener(subscriptionName = "my-subscription-1", topics = "my-topic-1")
public void listen2(Integer message) {
System.out.println(message);
}
----
====
The following `PulsarListener` method shows how we can consume complex types from a topic
====
[source, java]
----
@PulsarListener(subscriptionName = "my-subscription-2", topics = "my-topic-2", schemaType = SchemaType.JSON)
public void listen3(Foo message) {
System.out.println(message);
}
----
====
Note the addition of a `schemaType` property on `PulsarListener`.
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.
Here is an example of using `PulsarListener` to consume records in batches.
====
[source, java]
----
@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);
}
}
----
====
Note that in this example, we are receiving the records as a collection (`List`) of objects.
In addition, in order to enable batch consumption at the `PulsarListener` level, you need to set the `batch` property on the annotation to `true`.
Based on the actual type that the `List` holds, the framework tries to infer the schema to use.
If the `List` contains a complex type, then the `schemaType` still needs to be provided on `PulsarListener`.
The following also should work in which we use the `Messages` holder type provided by the Pulsar Java client.
====
[source, java]
----
@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);
}
}
----
====
[[pulsar-message-listener-container]]
=== Pulsar Message Listener Container
As briefly mentioned above, the message listener container is at the heart of message consumption when using Spring for Apache Pulsar.
`PulsarListener` uses the message listener container infrastructure behind the scenes to create and manage the Pulsar consumer.
Spring for Apache Pulsar provides the contract for this message listener container through `PulsarMessageListenerContainer`.
The default implementation for this message listener container is provided through `DefaultPulsarMessageListenerContainer`.
As its name indicates, `PulsarMessageListenerContainer` contains the message listener.
The container creates the Pulsar consumer and then runs a separate thread to receive and handle the data.
The data is handled by the provided message listener implementation.
The message listener container consumes the data in batch using the consumer's `batchReceive` method.
Once data is received, it is handed over to the selected message listener implementation.
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);
}
interface PulsarAcknowledgingMessageListener<T> extends PulsarRecordMessageListener<T> {
void received(Consumer<T> consumer, Message<T> msg, Acknowledgement acknowledgement);
}
interface PulsarBatchMessageListener<T> extends PulsarRecordMessageListener<T> {
void received(Consumer<T> consumer, Messages<T> msg);
}
interface PulsarBatchAcknowledgingMessageListener<T> extends PulsarBatchMessageListener<T> {
void received(Consumer<T> consumer, Messages<T> msg, Acknowledgement acknowledgement);
}
----
====
We will see the details about these various message listeners in the sections below.
=== Consuming the Records
In this section, we are going to see how the message listener container enables both single record and batch based message consumption.
==== Single Record Consumption
Let us re-visit our basic `PulsarListener` for the sake of this discussion.
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public void listen(String message) {
System.out.println("Message Received: " + message);
}
----
====
With this `PulsarListener` method, what we are essentially doing is that asking Spring for Apache Pulsar to invoke the listener method with a single record each time.
We mentioned that the message listener container consumes the data in batches using the `batchReceive` method on the consumer.
The framework detects that the `PulsarListener` in this case receives a single record which means that on each invocation of the method it needs a singe record.
Although the records are consumed by the message listener container in batches, it iterates through the received batch and then invoke the listener method through an adapter for `PulsarRecordMessageListener`.
As you can see in the previous section, `PulsarRecordMessageListener` simply extends from the `MessageListener` provided by the Pulsar Java client and it supports the basic `received` method.
==== Batch Consumption
Here is the `PulsarListener` example of consuming records in batches.
====
[source, java]
----
@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);
}
}
----
====
When using this type of `PulsarListener`, the framework detects that you are in batch mode.
Since it is already received the data in batches using the Consumer's `batchReceive` method, it simply hands off the entire batch to the listener method through an adapter for `PulsarBatchMessageListener`.
=== Message Acknowledgment
When using Spring for Apache Pulsar, the message acknowledgment is handled by the framework unless opted out by the application.
In this section, we go through the details of how the framework takes care of message acknowledgment.
==== Message ACK modes
Spring for Apache Pulsar provides the following modes for acknowledging messages
```
BATCH,
RECORD,
MANUAL;
```
`BATCH` acknowledgment mode is the default, but you can change it on the message listener container.
==== Message Ack in Single Record Mode
When consuming single records using `PulsarRecordMessageListener` and the default ack mode of `BATCH` is used, the framework waits for all the record received from the `batchReceive` call to process successfully and then call the `acknowledge` method on the Pulsar Consumer.
If any particular record throws an exception when invoking the handler method, Spring for Apache Pulsar tracks those records and separately call `negativeAcknowledge` on those records after the entire batch is processed.
If the application wants the acknowledgment of negative acknoweldgement to occur per record, then the `RECORD` ack mode can be enabled.
In that case, after handling each record the message is acknowledged if no error or negatively acknowledged if there was an error.
==== Message Ack in Batch Consumption
When records are consumed in batches (See the section above), then if the default ack mode of `BATCH` is used, then when the entire batch is processed successfully, it will be acknowledged.
If any records throw an exception, then the entire batch is negatively acknowledged.
When consuming in batch mode, `RECORD` is not an allowed ack mode.
This might cause an issue as application does not want the entire batch to be re-delivered again.
For such situations, you need to use the `MANUAL` acknowledgement mode.
==== Manual Acknowledgment
When `MANUAL` ack mode is set on the message listener container, then the framework will not do any acknowledgment - positive or negative.
It is entirely up to the application to take care of such concerns.
When `MANUAL` ack mode is set, Spring for Apache Pulsar selects a compatible message listener container - `PulsarAcknowledgingMessageListener` when in record consumption and `PulsarBatchAcknowledgingMessageListener` for batch consumption.
These interfaces provide you access to an `Acknowledgment` object.
The `Acknowledgment` object provides the following API methods.
====
[source, java]
----
void acknowledge();
void acknowledge(MessageId messageId);
void acknowledge(List<MessageId> messageIds);
void nack();
void nack(MessageId messageId);
----
====
You can inject this `Acknowledgment` object to your `PulsarListener` while using `MANUAL` ack mode and then call the corresponding method.
Here is a basic example for a record based listener.
====
[source, java]
----
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public void listen(String message, Acknowlegement acknowledgment) {
System.out.println("Message Received: " + message);
acknowledgment.acknowledge();
}
----
====
You can also call `acknowledgment.nack()` to negatively acknowledge in which case the record will be re-delivered.
When using a batch listener, the message listener container cannot know which record it is currently operating upon.
Therefore, in order to manually acknowledge, you need to use one of the overloaded `acknowledge` method that takes a `MessageId` or a `List<MessageId>`.
You can also negatively acknowledge with the `MessageId` for the batch listener.
=== Partitioned topics - Publishing and Consuming.
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.
```
@SpringBootApplication
public class PulsarBootPartitioned {
public static void main(String[] args) {
SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition");
}
@Bean
public ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
pulsarTemplate.setDefaultTopicName("hello-pulsar-partitioned");
return args -> {
for (int i = 0; i < 10; i++) {
pulsarTemplate.sendAsync("hello john doe 0 ", new FooRouter());
pulsarTemplate.sendAsync("hello alice doe 1", new BarRouter());
pulsarTemplate.sendAsync("hello buzz doe 2", new BuzzRouter());
}
};
}
@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned")
public void listen(String message) {
System.out.println("Message Received: " + message);
}
static class FooRouter implements MessageRouter {
@Override
public int choosePartition(Message<?> msg, TopicMetadata metadata) {
return 0;
}
}
static class BarRouter implements MessageRouter {
@Override
public int choosePartition(Message<?> msg, TopicMetadata metadata) {
return 1;
}
}
static class BuzzRouter implements MessageRouter {
@Override
public int choosePartition(Message<?> msg, TopicMetadata metadata) {
return 2;
}
}
}
```
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.
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.
What can we do if we want each partition to be consumed by a single distinct consumer?
We can switch to the `failover` subscription mode and add three separate consumers.
Here is an example.
```
@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover")
public void listen1(String foo) {
System.out.println("Message Received 1: " + foo);
}
@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover")
public void listen2(String foo) {
System.out.println("Message Received 2: " + foo);
}
@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover")
public void liste3n(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.
In the similar vein, if you want to use Pulsar's shared consumer type, you can use the subscription type `shared`.
Keep in mind though, that when using the `shared` mode, you lose any ordering guarantees as a single consumer may receive messages from all the partitions before another consumer gets a chance.
Here is an example.
```
@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "shared")
public void listen1(String foo) {
System.out.println("Message Received 1: " + foo);
}
@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "shared")
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.
```
@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.
```
@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.
```
@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.
=== Specify schema information
As indicated above, for normal Java types (the primitive ones), Spring Pulsar framework can infer the proper Schema to use on the `PulsarListener`.
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.
```
@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`.

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

@@ -0,0 +1,105 @@
[[quick-tour]]
=== Quick Tour
In this section, we will take a quick tour of Spring for Apache Pulsar.
#### Minimum Supported Versions
The following are the minimum supported versions of the underlying libraries required by Spring for Apache Pulsar
**JDK** - JDK 17
**Apache Pulsar** - 2.10.0
**Spring Boot** - 3.0.0
**Spring Framework** - 6.0.0
### Building the Project
Spring for Apache Pulsar uses gradle as it's build tool.
Use the following command to do a full build of the project.
```
./gradlew clean build
```
The build will produce the following artifacts.
* spring-pulsar
* spring-pulsar-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
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`.
```
@SpringBootApplication
public class PulsarBootHelloWorld {
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));
}
};
}
@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar")
public 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.
In the sample above, we are heavily relying on Spring Boot auto-configuration.
Spring Boot auto-configures several components for our application.
It automatically provides a `PulsarClient` for the application which is used by both the producer and the consumer.
`PulsarTemplate` also is auto-configured by Spring Boot which we inject in the application and start sending records to a Pulsar topic.
The application sends messages to a topic named `hello-pulsar`.
Note that the application does not specify any schema information.
That is because Spring for Apache Pulsar library automatically infers the schema type from the type of the data that you are sending.
We use `PulsarListener` annotation to consume from the `hello-pulsar` topic where we publish the data.
`PulsarListener` is a convenient annotation that wraps the message listener container infrastructure in Spring for Apache Pulsar.
Behind the scenes, it creates a message listener container which creates and manages the Pulsar consumer.
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.
Here also, the framework infers the schema type used from the data type that the `PulsarListner` method uses as the payload - `String` in this case.