diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/nav.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/nav.adoc index 803e0467..8922d4d4 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/nav.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/nav.adoc @@ -6,6 +6,12 @@ ** xref:intro/getting-help.adoc[] * xref:reference/reference.adoc[] ** xref:reference/pulsar.adoc[] +*** xref:reference/pulsar/quick-tour.adoc[] +*** xref:reference/pulsar/pulsar-client.adoc[] +*** xref:reference/pulsar/message-production.adoc[] +*** xref:reference/pulsar/message-consumption.adoc[] +*** xref:reference/pulsar/topic-resolution.adoc[] +*** xref:reference/pulsar/publishing-consuming-partitioned-topics.adoc[] *** xref:reference/tombstones.adoc[] ** xref:reference/reactive-pulsar.adoc[] *** xref:reference/tombstones-reactive.adoc[] diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc index 4c47edff..3f8f5f06 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc @@ -18,7 +18,7 @@ See the {spring-boot-pulsar-config-props}[`spring.pulsar.admin.*`] application p [[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 xref:reference/pulsar.adoc#client-authentication[security configuration] by replacing `spring.pulsar.client` with `spring.pulsar.admin`. +You can use the aforementioned xref:reference/pulsar/pulsar-client.adoc#client-authentication[security configuration] by replacing `spring.pulsar.client` with `spring.pulsar.admin`. [[pulsar-auto-topic-creation]] == Automatic Topic Creation diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc index 4ec66c12..c6524281 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc @@ -1,1360 +1,5 @@ [[pulsar]] = Using Spring for Apache Pulsar -include::../attributes/attributes.adoc[] +:page-section-summary-toc: 1 -== Preface - -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-spring-boot-starter` module as a dependency. - -NOTE: The majority of this reference expects the reader to be using the starter and gives most directions for configuration with that in mind. -However, an effort is made to call out when instructions are specific to the Spring Boot starter usage. - -include::quick-tour.adoc[leveloffset=+1] - -[[pulsar-client]] -== Pulsar Client - -When you use the Pulsar Spring Boot Starter, you get the `PulsarClient` auto-configured. - -By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`. -This can be adjusted by setting the `spring.pulsar.client.service-url` property to a different value. - -TIP: The value must be a valid {apache-pulsar-docs}/client-libraries-java/#connection-urls[Pulsar Protocol] URL - -You can further configure the client by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application properties. - -NOTE: If you are not using the starter, you will need to configure and register the `PulsarClient` yourself. -There is a `DefaultPulsarClientFactory` that accepts a builder customizer that can be used to help with this. - -[[tls-encryption]] -=== TLS Encryption (SSL) -include::tls-encryption.adoc[] - -[[client-authentication]] -=== Authentication -include::authentication.adoc[] - -== Message Production - -[[pulsar-producer]] -=== Pulsar Template - -On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides methods to publish records through its contract. - -There are two categories of these send API methods: `send` and `sendAsync`. -The `send` methods block calls by 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`, which you can use to asynchronously receive the message ID once the messages are published. - -NOTE: For the API variants that do not include a topic parameter, a <> is used to determine the destination topic. - -==== Simple API -The template provides a handful of methods ({javadocs}/org/springframework/pulsar/core/PulsarOperations.html[prefixed with _'send'_]) for simple send requests. For more complicated send requests, a fluent API lets you configure more options. - -==== Fluent API -The template provides a {javadocs}/org/springframework/pulsar/core/PulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests. - -==== Message customization -You can specify a `TypedMessageBuilderCustomizer` to configure the outgoing message. For example, the following code shows how to send a keyed message: -[source, java] ----- -template.newMessage(msg) - .withMessageCustomizer((mb) -> mb.key("foo-msg-key")) - .send(); ----- - -[[single-producer-customize]] -==== Producer customization -You can specify a `ProducerBuilderCustomizer` to configure the underlying Pulsar producer builder that ultimately constructs the producer used to send the outgoing message. - -WARNING: Use with caution as this gives full access to the producer 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) - .withProducerCustomizer((pb) -> pb.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 `Producer` builder such as: -[source, java] ----- -template.newMessage(msg) - .withProducerCustomizer((pb) -> pb.messageRouter(messageRouter)) - .send(); ----- - -TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`. - -This other example shows how to add a `ProducerInterceptor` that will intercept and mutate messages received by the producer before being published to the brokers: -[source, java] ----- -template.newMessage(msg) - .withProducerCustomizer((pb) -> pb.intercept(interceptor)) - .send(); ----- - -The customizer will only apply to the producer used for the send operation. -If you want to apply a customizer to all producers, you must provide them to the producer factory as described in <>. - -CAUTION: The rules described in "`<>`" must be followed when using Lambda customizers. - - -[[schema-info-template-imperative]] -:template-class: PulsarTemplate -include::schema-info/schema-info-template.adoc[leveloffset=+1] - -[[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 which you can further configure by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.*`] application properties. - -NOTE: If topic information is not specified when using the producer factory APIs directly, the same <> used by the `PulsarTemplate` is used with the one exception that the "Message type default" step is **omitted**. - -[[global-producer-customize]] -==== Global producer customization -The framework provides the `ProducerBuilderCustomizer` contract which allows you to configure the underlying builder which is used to construct each producer. -To customize all producers, you can pass a list of customizers into the `PulsarProducerFactory` constructor. -When using multiple customizers, they are applied in the order in which they appear in the list. - -TIP: If you use Spring Boot auto-configuration, you can specify the customizers as beans and they will be passed automatically to the `PulsarProducerFactory`, ordered according to their `@Order` annotation. - -If you want to apply a customizer to just a single producer, you can use the Fluent API and <>. - -[[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-boot-pulsar-config-props}[`spring.pulsar.producer.cache.*`] application properties. - -[[producer-caching-lambdas]] -==== Caution on Lambda customizers -Any user-provided producer customizers are also included in the cache key. -Because the cache key relies on a valid implementation of `equals/hashCode`, one must take caution when using Lambda customizers. - -IMPORTANT: *RULE:* Two customizers implemented as Lambdas will match on `equals/hashCode` *if and only if* they use the same Lambda instance and do not require any variable defined outside its closure. - -To clarify the above rule we will look at a few examples. -In the following example, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. Additionally, it requires no variable outside its closure. Therefore, it *will* match as a cache key. - -[source, java] ----- -void sendUser() { - var user = randomUser(); - template.newMessage(user) - .withTopic("user-topic") - .withProducerCustomizer((b) -> b.producerName("user")) - .send(); -} ----- - -In this next case, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. However, it requires a variable outside its closure. Therefore, it *will not* match as a cache key. - -[source, java] ----- -void sendUser() { - var user = randomUser(); - var name = randomName(); - template.newMessage(user) - .withTopic("user-topic") - .withProducerCustomizer((b) -> b.producerName(name)) - .send(); -} ----- - -In this final example, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. While it does use a variable name, it does not originate outside its closure and therefore *will* match as a cache key. -This illustrates that variables can be used *within* the Lambda closure and can even make calls to static methods. - -[source, java] ----- -void sendUser() { - var user = randomUser(); - template.newMessage(user) - .withTopic("user-topic") - .withProducerCustomizer((b) -> { - var name = SomeHelper.someStaticMethod(); - b.producerName(name); - }) - .send(); -} ----- - -IMPORTANT: *RULE:* If your Lambda customizer is not defined *once and only once* (the same instance is used on subsequent calls) *OR* it requires variable(s) defined outside its closure then you must provide a customizer implementation with a valid `equals/hashCode` implementation. - -WARNING: If these rules are not followed then the producer cache will always miss and your application performance will be negatively affected. - -=== Intercept Messages on the Producer -Adding a `ProducerInterceptor` lets you intercept and mutate messages received by the producer before they are published to the brokers. -To do so, you can pass a list of interceptors into the `PulsarTemplate` constructor. -When using multiple interceptors, the order they are applied in is the order in which they appear in the list. - -If you use Spring Boot auto-configuration, you can specify the interceptors as Beans. -They are passed automatically to the `PulsarTemplate`. -Ordering of the interceptors is achieved by using the `@Order` annotation as follows: - -[source, java] ----- -@Bean -@Order(100) -ProducerInterceptor firstInterceptor() { - ... -} - -@Bean -@Order(200) -ProducerInterceptor secondInterceptor() { - ... -} ----- - -NOTE: If you are not using the starter, you will need to configure and register the aforementioned components yourself. - - -== Message Consumption - -[[pulsar-listener]] -=== Pulsar Listener - -When it comes to Pulsar consumers, we recommend that end-user applications use the `PulsarListener` annotation. -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 the underlying Pulsar consumer that it uses to consume messages. - -Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties. -**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**: - -TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation. - -TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default. - - -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); -} ----- - -You can further simplify this method: - -[source, java] ----- -@PulsarListener -public void listen(String message) { - System.out.println("Message Received: " + message); -} ----- - -In this most basic form, when the `subscriptionName` is not provided on the `@PulsarListener` annotation an auto-generated subscription name will be used. -Likewise, when the `topics` are not directly provided, a <> is used to determine the destination topic. - -In the `PulsarListener` 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 and provides that schema to the consumer. -The framework does this inference for all primitive types. -For all non-primitive types the default schema is assumed to be JSON. -If a complex type is using anything besides JSON (such as AVRO or KEY_VALUE) you must provide the schema type on the annotation using the `schemaType` property. - -The following example shows another `PulsarListener` method, which takes an `Integer`: - -[source, java] ----- -@PulsarListener(subscriptionName = "my-subscription-1", topics = "my-topic-1") -public void listen(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 listen(Foo message) { - System.out.println(message); -} ----- - -Let us look at a few more ways. - -You can consume the Pulsar message directly: - -[source, java] ----- -@PulsarListener(subscriptionName = "my-subscription", topics = "my-topic") -public void listen(org.apache.pulsar.client.api.Message message) { - System.out.println(message.getValue()); -} ----- - -The following example consumes the record by using the Spring messaging envelope: - -[source, java] ----- -@PulsarListener(subscriptionName = "my-subscription", topics = "my-topic") -public void listen(org.springframework.messaging.Message message) { - System.out.println(message.getPayload()); -} ----- - -Now let us see how we can consume records in batches. -The following example uses `PulsarListener` to consume records in batches as POJOs: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) -public void listen(List messages) { - System.out.println("records received :" + messages.size()); - messages.forEach((message) -> System.out.println("record : " + message)); -} ----- - -Note that, in this example, we receive the records as a collection (`List`) of objects. -In addition, 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 besides JSON, you still need to provide the `schemaType` on `PulsarListener`. - -The following uses the `Message` envelope provided by the Pulsar Java client: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) -public void listen(List> messages) { - System.out.println("records received :" + messages.size()); - messages.forEach((message) -> System.out.println("record : " + message.getValue())); -} ----- - -The following example consumes batch records with an envelope of the Spring messaging `Message` type: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) -public void listen(List> messages) { - System.out.println("records received :" + messages.size()); - messages.forEach((message) -> System.out.println("record : " + message.getPayload())); -} ----- - -Finally, you can also use the `Messages` holder object from Pulsar for the batch listener: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) -public void listen(org.apache.pulsar.client.api.Messages> messages) { - System.out.println("records received :" + messages.size()); - messages.forEach((message) -> System.out.println("record : " + message.getValue())); -} ----- - -When you use `PulsarListener`, you can provide Pulsar consumer properties directly on the annotation itself. -This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `PulsarListener` methods. - -The following example uses Pulsar consumer properties directly on `PulsarListener`: - -[source, java] ----- -@PulsarListener(properties = { "subscriptionName=subscription-1", "topicNames=foo-1", "receiverQueueSize=5000" }) -void listen(String message) { -} ----- - -TIP: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` application configuration properties - -[[listener-auto-consume]] -==== Generic records with AUTO_CONSUME -If there is no chance to know the type of schema of a Pulsar topic in advance, you can use the `AUTO_CONSUME` schema type to consume generic records. -In this case, the topic deserializes messages into `GenericRecord` objects using the schema info associated with the topic. - -To consume generic records set the `schemaType = SchemaType.AUTO_CONSUME` on your `@PulsarListener` and use a Pulsar message of type `GenericRecord` as the message parameter as shown below. - -[source, java] ----- -@PulsarListener(topics = "my-generic-topic", schemaType = SchemaType.AUTO_CONSUME) -void listen(org.apache.pulsar.client.api.Message message) { - GenericRecord record = message.getValue(); - record.getFields().forEach((f) -> - System.out.printf("%s = %s%n", f.getName(), record.getField(f))); -} ----- - -TIP: The `GenericRecord` API allows access to the fields and their associated values - -==== Customizing the ConsumerBuilder - -You can customize any fields available through `ConsumerBuilder` using a `PulsarListenerConsumerBuilderCustomizer` by providing a `@Bean` of type `PulsarListenerConsumerBuilderCustomizer` and then making it available to the `PulsarListener` as shown below. - -[source, java] ----- -@PulsarListener(topics = "hello-topic", consumerCustomizer = "myCustomizer") -public void listen(String message) { - System.out.println("Message Received: " + message); -} - -@Bean -PulsarListenerConsumerBuilderCustomizer myCustomizer() { - return (builder) -> builder.consumerName("myConsumer"); -} ----- - -TIP: If your application only has a single `@PulsarListener` and a single `PulsarListenerConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied. - - -[[schema-info-listener-imperative]] -:listener-class: PulsarListener -include::schema-info/schema-info-listener.adoc[leveloffset=+1] - -=== Accessing the Pulsar Consumer Object -Sometimes, you need direct access to the Pulsar Consumer object. -The following example shows how to get it: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") -public void listen(String message, org.apache.pulsar.client.api.Consumer consumer) { - System.out.println("Message Received: " + message); - ConsumerStats stats = consumer.getStats(); - ... -} ----- - -CAUTION: When accessing the `Consumer` object this way, do NOT 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. - -[[pulsar-message-listener-container]] -=== Pulsar Message Listener Container - -Now that we saw the basic interactions on the consumer side through `PulsarListener`. Let us now dive into the inner workings of how `PulsarListener` interacts with the underlying Pulsar consumer. -Keep in mind that, for end-user applications, in most scenarios, we recommend using the `PulsarListener` annotation directly for consuming from a Pulsar topic when using Spring for Apache Pulsar, as that model covers a broad set of application use cases. -However, it is important to understand how `PulsarListener` works internally. This section goes through those details. - -As briefly mentioned earlier, the message listener container is at the heart of message consumption when you use 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 by 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 you use Spring for Apache Pulsar. - -* link:{github}/blob/8e33ac0b122bc0e75df299919c956cacabcc9809/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java#L29[PulsarRecordMessageListener] - -* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarAcknowledgingMessageListener.java#L28[PulsarAcknowledgingMessageListener] - -* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java#L36[PulsarBatchMessageListener] - -* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java#L28[PulsarBatchAcknowledgingMessageListener] - -We see the details about these various message listeners in the following sections. - -Before doing so, however, let us take a closer look at the container itself. - -==== DefaultPulsarMessageListenerContainer - -This is a single consumer-based message listener container. -The following listing shows its constructor: - -[source, java] ----- -public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, - PulsarContainerProperties pulsarContainerProperties) -} ----- - -It receives a `PulsarConsumerFactory` (which it uses to create the consumer) and a `PulsarContainerProperties` object (which contains information about the container properties). -`PulsarContainerProperties` has the following constructors: - -[source, java] ----- -public PulsarContainerProperties(String... topics) - -public PulsarContainerProperties(Pattern topicPattern) ----- - -You can provide the topic information through `PulsarContainerProperties` or as a consumer property that is provided to the consumer factory. -The following example uses the `DefaultPulsarMessageListenerContainer`: - -[source, java] ----- -Map config = new HashMap<>(); -config.put("topics", "my-topic"); -PulsarConsumerFactory pulsarConsumerFactorY = DefaultPulsarConsumerFactory<>(pulsarClient, config); - -PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); - -pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - }); - -DefaultPulsarMessageListenerContainer pulsarListenerContainer = new DefaultPulsarMessageListenerContainer(pulsarConsumerFacotyr, - pulsarContainerProperties); - -return pulsarListenerContainer; ----- - -NOTE: If topic information is not specified when using the listener containers directly, the same <> used by the `PulsarListener` is used with the one exception that the "Message type default" step is **omitted**. - -`DefaultPulsarMessageListenerContainer` creates only a single consumer. -If you want to have multiple consumers managed through multiple threads, you need to use `ConcurrentPulsarMessageListenerContainer`. - -==== ConcurrentPulsarMessageListenerContainer - -`ConcurrentPulsarMessageListenerContainer` has the following constructor: - -[source, java] ----- -public ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, - PulsarContainerProperties pulsarContainerProperties) ----- - -`ConcurrentPulsarMessageListenerContainer` lets you specify a `concurrency` property through a setter. -Concurrency of more than `1` is allowed only on non-exclusive subscriptions (`failover`, `shared`, and `key-shared`). -You can only have the default `1` for concurrency when you have an exclusive subscription mode. - -The following example enables `concurrency` through the `PulsarListener` annotation for a `failover` subscription. - -[source, java] ----- -@PulsarListener(topics = "my-topic", subscriptionName = "subscription-1", - subscriptionType = SubscriptionType.Failover, concurrency = "3") -void listen(String message, Consumer consumer) { - ... - System.out.println("Current Thread: " + Thread.currentThread().getName()); - System.out.println("Current Consumer: " + consumer.getConsumerName()); -} ----- - -In the preceding listener, it is assumed that the topic `my-topic` has three partitions. -If it is a non-partitioned topic, having concurrency set to `3` does nothing. You get two idle consumers in addition to the main active one. -If the topic has more than three partitions, messages are load-balanced across the consumers that the container creates. -If you run this `PulsarListener`, you see that messages from different partitions are consumed through different consumers, as implied by the thread name and consumer names printouts in the preceding example. - -NOTE: When you use the `Failover` subscription this way on partitioned topics, Pulsar guarantees message ordering. - -The following listing shows another example of `PulsarListener`, but with `Shared` subscription and `concurrency` enabled. - -[source, java] ----- -@PulsarListener(topics = "my-topic", subscriptionName = "subscription-1", - subscriptionType = SubscriptionType.Shared, concurrency = "5") -void listen(String message) { - ... -} ----- - -In the preceding example, the `PulsarListener` creates five different consumers (this time, we assume that the topic has five partitions). - -NOTE: In this version, there is no message ordering, as `Shared` subscriptions do not guarantee any message ordering in Pulsar. - -If you need message ordering and still want a shared subscription types, you need to use the `Key_Shared` subscription type. - -==== Message Consumption - -Let us take a look at how the message listener container enables both single-record and batch-based message consumption. - -[discrete] -==== Single Record Consumption -Let us revisit 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, we essential ask 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. This 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 invokes the listener method through an adapter for `PulsarRecordMessageListener`. -As you can see in the previous section, `PulsarRecordMessageListener` extends from the `MessageListener` provided by the Pulsar Java client, and it supports the basic `received` method. - -[discrete] -==== Batch Consumption -The following example shows the `PulsarListener` consuming records in batches: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) -public void listen4(List messages) { - System.out.println("records received :" + messages.size()); - messages.forEach((message) -> System.out.println("record : " + message)); -} ----- - -When you use this type of `PulsarListener`, the framework detects that you are in batch mode. -Since it already received the data in batches by using the Consumer's `batchReceive` method, it hands off the entire batch to the listener method through an adapter for `PulsarBatchMessageListener`. - -[[pulsar-headers]] -=== Pulsar Headers -The Pulsar message metadata can be consumed as Spring message headers. -The list of available headers can be found in {github}/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java[PulsarHeaders.java]. - -==== Accessing in Single Record based Consumer - -The following example shows how you can access the various Pulsar Headers in an application that uses the single record mode of consuming: - -[source,java] ----- -@PulsarListener(topics = "simpleListenerWithHeaders") -void simpleListenerWithHeaders(String data, @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, - @Header(PulsarHeaders.RAW_DATA) byte[] rawData, - @Header("foo") String foo) { - -} ----- - -In the preceding example, we access the values for the `messageId` and `rawData` 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 Batch Record based Consumer - -In this section, we see how to access the various Pulsar Headers in an application that uses a batch consumer: - -[source,java] ----- -@PulsarListener(topics = "simpleBatchListenerWithHeaders", batch = true) -void simpleBatchListenerWithHeaders(List data, - @Header(PulsarHeaders.MESSAGE_ID) List messageIds, - @Header(PulsarHeaders.TOPIC_NAME) List topicNames, @Header("foo") List fooValues) { - -} ----- - -In the preceding example, we consume the data as a `List`. -When extracting the various headers, we do so as a `List<>` as well. -Spring for Apache 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.Messages`, or `org.springframework.messaging.Messsge`. - -=== Message Acknowledgment - -When you use 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]] -==== 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. -In the following sections, we see how acknowledgment works when you use both single and batch versions of `PulsarListener` and how they translate to the backing message listener container (and, ultimately, to the Pulsar consumer). - -==== Automatic Message Ack in Single Record Mode - -Let us revisit our basic single message based `PulsarListener`: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") -public void listen(String message) { - System.out.println("Message Received: " + message); -} ----- - -It is natural to wonder, how acknowledgment works when you use `PulsarListener`, especially if you are familiar with using Pulsar consumer directly. -The answer comes down to the message listener container, as that is the central place in Spring for Apache Pulsar that coordinates all the consumer related activities. - -Assuming you are not overriding the default behavior, this is what happens behind the scenes when you use the preceding `PulsarListener`: - -. First, the listener container receives messages as batches from the Pulsar consumer. -. The received messages are handed down to `PulsarListener` one message at a time. -. When all the records are handed down to the listener method and successfully processed, the container acknowledges all the messages from the original batch. - -This is the normal flow. If any records from the original batch throw an exception, Spring for Apache Pulsar track those records separately. -When all the records from the batch are processed, Spring for Apache Pulsar acknowledges all the successful messages and negatively acknowledges (nack) all the failed messages. -In other words, when consuming single records by using `PulsarRecordMessageListener` and the default ack mode of `BATCH` is used, the framework waits for all the records received from the `batchReceive` call to process successfully and then calls 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 calls `negativeAcknowledge` on those records after the entire batch is processed. - -If the application wants the acknowledgment or negative acknowledgment to occur per record, the `RECORD` ack mode can be enabled. -In that case, after handling each record, the message is acknowledged if no error and negatively acknowledged if there was an error. -The following example enables `RECORD` ack mode on the Pulsar listener: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", ackMode = AckMode.RECORD) -public void listen(String message) { - System.out.println("Message Received: " + message); -} ----- - -==== Manual Message Ack in Single Record Mode - -You might not always want the framework to send acknowledgments but, rather, do that directly from the application itself. -Spring for Apache Pulsar provides a couple of ways to enable manual message acknowledgments. The following example shows one of them: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", ackMode = AckMode.MANUAL) -public void listen(Message message, Acknowledgment acknowledgment) { - System.out.println("Message Received: " + message.getValue()); - acknowledgment.acknowledge(); -} ----- - -A few things merit explanation here. First, we enablE manual ack mode by setting `ackMode` on `PulsarListener`. -When enabling manual ack mode, Spring for Apache Pulsar lets the application inject an `Acknowledgment` object. -The framework achieves this by selecting a compatible message listener container: `PulsarAcknowledgingMessageListener` for single record based consumption, which gives 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 messageIds); - -void nack(); - -void nack(MessageId messageId); ----- - -You can inject this `Acknowledgment` object into your `PulsarListener` while using `MANUAL` ack mode and then call one of the corresponding methods. - -In the preceding `PulsarListener` example, we call a parameter-less `acknowledge` method. -This is because the framework knows which `Message` it is currently operating under. -When calling `acknowledge()`, you need not receive the payload with the `Message` enveloper` but, rather, use the target type -- `String`, in this example. -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. - -You can also call `acknowledge` directly on the Pulsar consumer: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", ackMode = AckMode.MANUAL) -public void listen(Message message, Consumer consumer) { - System.out.println("Message Received: " + message.getValue()); - try { - consumer.acknowledge(message); - } - catch (Exception e) { - .... - } -} ----- - -When calling `acknowledge` directly on the underlying consumer, you need to do error handling by yourself. -Using the `Acknowledgment` does not require that, as the framework can do that for you. -Therefore, you should use the `Acknowledgment` object approach when using manual acknowledgment. - -IMPORTANT: When using manual acknowledgment, it is important to understand that the framework completely stays from any acknowledgment at all. -Hence, it is extremely important to think through the right acknowledgment strategies when designing applications. - -==== Automatic Message Ack in Batch Consumption - -When you consume records in batches (see "`<>`") and you use the default ack mode of `BATCH` is used, when the entire batch is processed successfully, the entire batch is acknowledged. -If any records throw an exception, the entire batch is negatively acknowledged. -Note that this may not be the same batch that was batched on the producer side. Rather, this is the batch that returned from calling `batchReceive` on the consumer - -Consider the following batch listener: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", batch = true) -public void batchListen(List messages) { - for (Foo foo : messages) { - ... - } -} ----- - -When all the messages in the incoming collection (`messages` in this example) are processed, the framework acknowledges all of them. - -When consuming in batch mode, `RECORD` is not an allowed ack mode. -This might cause an issue, as an application may not want the entire batch to be re-delivered again. -In such situations, you need to use the `MANUAL` acknowledgement mode. - -==== Manual Message Ack in Batch Consumption - -As seen in the previous section, when `MANUAL` ack mode is set on the message listener container, the framework does 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: `PulsarBatchAcknowledgingMessageListener` for batch consumption, which gives you access to an `Acknowledgment` object. -The following are the methods available in the `Acknowledgment` API: - -[source, java] ----- -void acknowledge(); - -void acknowledge(MessageId messageId); - -void acknowledge(List messageIds); - -void nack(); - -void nack(MessageId messageId); ----- - -You can inject this `Acknowledgment` object into your `PulsarListener` while using `MANUAL` ack mode. -The following listing shows a basic example for a batch based listener: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") -public void listen(List> messgaes, Acknowlegement acknowledgment) { - for (Message message : messages) { - try { - ... - acknowledgment.acknowledge(message.getMessageId()); - } - catch (Exception e) { - acknowledgment.nack(message.getMessageId()); - } - } -} ----- - -When you use a batch listener, the message listener container cannot know which record it is currently operating upon. -Therefore, to manually acknowledge, you need to use one of the overloaded `acknowledge` method that takes a `MessageId` or a `List`. -You can also negatively acknowledge with the `MessageId` for the batch listener. - -=== Message Redelivery and Error Handling - -Now that we have seen both `PulsarListener` and the message listener container infrastructure and its various functions, let us now try to understand message redelivery and error handling. -Apache Pulsar provides various native strategies for message redelivery and error handling. We take a look at them and see how we can use them through Spring for Apache Pulsar. - -==== Specifying Acknowledgment Timeout for Message Redelivery - -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. -If the ack timeout property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered. - -When you use Spring for Apache Pulsar, you can set this property via a <<_consumer_customization_on_pulsarlistener,consumer customizer>> or with the native Pulsar `ackTimeout` property in the `properties` attribute of `@PulsarListener`: - -[source, java] ----- -@PulsarListener(subscriptionName = "subscription-1", topics = "topic-1" - properties = {"ackTimeout=60s"}) -public void listen(String s) { - ... -} ----- - -When you specify the ack timeout, if the consumer does not send an acknowledgement within 60 seconds, the message is redelivered by Pulsar to the consumer. - -If you want to specify some advanced backoff options for ack timeout with different delays, you can do the following: - -[source, java] ----- -@EnablePulsar -@Configuration -class AckTimeoutRedeliveryConfig { - - @PulsarListener(subscriptionName = "withAckTimeoutRedeliveryBackoffSubscription", - topics = "withAckTimeoutRedeliveryBackoff-test-topic", - ackTimeoutRedeliveryBackoff = "ackTimeoutRedeliveryBackoff", - properties = { "ackTimeout=60s" }) - void listen(String msg) { - // some long-running process that may cause an ack timeout - } - - @Bean - RedeliveryBackoff ackTimeoutRedeliveryBackoff() { - return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(10 * 1000).multiplier(2) - .build(); - } - -} ----- - -In the preceding example, we specify a bean for Pulsar's `RedeliveryBackoff` with a minimum delay of 1 second, a maximum delay of 10 seconds, and a backoff multiplier of 2. -After the initial ack timeout occurs, the message redeliveries are controlled through this backoff bean. -We provide the backoff bean to the `PulsarListener` annotation by setting the `ackTimeoutRedeliveryBackoff` property to the actual bean name -- `ackTimeoutRedeliveryBackoff`, in this case. - -==== Specifying Negative Acknowledgment Redelivery - -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 via a <<_consumer_customization_on_pulsarlistener,consumer customizer>> or with the native Pulsar `negativeAckRedeliveryDelay` property in the `properties` attribute of `@PulsarListener`: - -[source, java] ----- -@PulsarListener(subscriptionName = "subscription-1", topics = "topic-1" - properties = {"negativeAckRedeliveryDelay=10ms"}) -public void listen(String s) { - ... -} ----- - -You can also specify different delays and backoff mechanisms with a multiplier by providing a `RedeliveryBackoff` bean and providing the bean name as the `negativeAckRedeliveryBackoff` property on the PulsarProducer, as follows: - -[source, java] ----- -@EnablePulsar -@Configuration -class NegativeAckRedeliveryConfig { - - @PulsarListener(subscriptionName = "withNegRedeliveryBackoffSubscription", - topics = "withNegRedeliveryBackoff-test-topic", negativeAckRedeliveryBackoff = "redeliveryBackoff", - subscriptionType = SubscriptionType.Shared) - void listen(String msg) { - throw new RuntimeException("fail " + msg); - } - - @Bean - RedeliveryBackoff redeliveryBackoff() { - return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(10 * 1000).multiplier(2) - .build(); - } - -} ----- - -==== Using Dead Letter Topic from Apache Pulsar for Message Redelivery and Error Handling - -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] ----- -@EnablePulsar -@Configuration -class DeadLetterPolicyConfig { - - @PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription", - topics = "topic-with-dlp", deadLetterPolicy = "deadLetterPolicy", - subscriptionType = SubscriptionType.Shared, properties = { "ackTimeout=1s" }) - void listen(String msg) { - throw new RuntimeException("fail " + msg); - } - - @PulsarListener(id = "dlqListener", topics = "my-dlq-topic") - void listenDlq(String msg) { - System.out.println("From DLQ: " + msg); - } - - @Bean - DeadLetterPolicy deadLetterPolicy() { - return DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic("my-dlq-topic").build(); - } - -} ----- - -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 `--DLQ` in Pulsar. -Next, we provide this bean name to `PulsarListener` by setting the `deadLetterPolicy` property. -Note that the `PulsarListener` 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 DLQ topic. -We have another `PulsarListener` 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-`, 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-` 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. -**** - -==== Native Error Handling in Spring for Apache Pulsar - -As we noted earlier, the DLQ feature in Apache Pulsar works only for shared subscriptions. -What does an application do if it needs to use some similar feature for non-shared subscriptions? -The main reason Pulsar does not support DLQ on exclusive and failover subscriptions is because those subscription types are order-guaranteed. -Allowing redeliveries, DLQ, and so on effectively receives messages out of order. -However, what if an application are okay with that but, more importantly, needs this DLQ feature for non-shared subscriptions? -For that, Spring for Apache Pulsar provides a `PulsarConsumerErrorHandler`, which you can use across any subscription types in Pulsar: `Exclusive`, `Failover`, `Shared`, or `Key_Shared`. - -When you use `PulsarConsumerErrorHandler` from Spring for Apache Pulsar, make sure not to set the ack timeout properties on the listener. - -Let us see some details by examining a few code snippets: - -[source, java] ----- -@EnablePulsar -@Configuration -class PulsarConsumerErrorHandlerConfig { - - @Bean - PulsarConsumerErrorHandler pulsarConsumerErrorHandler( - PulsarTemplate pulsarTemplate) { - return new DefaultPulsarConsumerErrorHandler<>( - new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, (c, m) -> "my-foo-dlt"), new FixedBackOff(100, 10)); - } - - @PulsarListener(id = "pulsarConsumerErrorHandler-id", subscriptionName = "pulsatConsumerErrorHandler-subscription", - topics = "pulsarConsumerErrorHandler-topic", - pulsarConsumerErrorHandler = "pulsarConsumerErrorHandler") - void listen(String msg) { - throw new RuntimeException("fail " + msg); - } - - @PulsarListener(id = "pceh-dltListener", topics = "my-foo-dlt") - void listenDlt(String msg) { - System.out.println("From DLT: " + msg); - } - -} ----- - -Consider the `pulsarConsumerErrorHandler` bean. -This creates a bean of type `PulsarConsumerErrorHandler` and uses the default implementation provided out of the box by Spring for Apache Pulsar: `DefaultPulsarConsumerErrorHandler`. -`DefaultPulsarConsumerErrorHandler` has a constructor that takes a `PulsarMessageRecovererFactory` and a `org.springframework.util.backoff.Backoff`. -`PulsarMessageRecovererFactory` is a functional interface with the following API: - -[source, java] ----- -@FunctionalInterface -public interface PulsarMessageRecovererFactory { - - /** - * Provides a message recoverer {@link PulsarMessageRecoverer}. - * @param consumer Pulsar consumer - * @return {@link PulsarMessageRecoverer}. - */ - PulsarMessageRecoverer recovererForConsumer(Consumer consumer); - -} - ----- - -The `recovererForConsumer` method takes a Pulsar consumer and returns a `PulsarMessageRecoverer`, which is another functional interface. -Here is the API of `PulsarMessageRecoverer`: - -[source, java] ----- -public interface PulsarMessageRecoverer { - - /** - * Recover a failed message, for e.g. send the message to a DLT. - * @param message Pulsar message - * @param exception exception from failed message - */ - void recoverMessage(Message message, Exception exception); - -} ----- - -Spring for Apache Pulsar provides an implementation for `PulsarMessageRecovererFactory` called `PulsarDeadLetterPublishingRecoverer` that provides a default implementation that can recover the message by sending it to a Dead Letter Topic (DLT). -We provide this implementation to the constructor for the preceding `DefaultPulsarConsumerErrorHandler`. -As the second argument, we provide a `FixedBackOff`. -You can also provide the `ExponentialBackoff` from Spring for advanced backoff features. -Then we provide this bean name for the `PulsarConsumerErrorHandler` as a property to the `PulsarListener`. -The property is called `pulsarConsumerErrorHandler`. -Each time the `PulsarListener` method fails for a message, it gets retried. -The number of retries are controlled by the `Backoff` provided implementation values. In our example, we do 10 retries (11 total tries -- the first one and then the 10 retries). -Once all the retries are exhausted, the message is sent to the DLT topic. - -The `PulsarDeadLetterPublishingRecoverer` implementation we provide uses a `PulsarTemplate` that is used for publishing the message to the DLT. -In most cases, the same auto-configured `PulsarTemplate` from Spring Boot is sufficient with the caveat for partitioned topics. -When using partitioned topics and using custom message routing for the main topic, you must use a different `PulsarTemplate` that does not take the auto-configured `PulsarProducerFactory` that is populated with a value of `custompartition` for `message-routing-mode`. -You can use a `PulsarConsumerErrorHandler` with the following blueprint: - -[source, java] ----- -@Bean -PulsarConsumerErrorHandler pulsarConsumerErrorHandler(PulsarClient pulsarClient) { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Map.of()); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); - - BiFunction, Message, String> destinationResolver = - (c, m) -> "my-foo-dlt"; - - PulsarDeadLetterPublishingRecoverer pulsarDeadLetterPublishingRecoverer = - new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, destinationResolver); - - return new DefaultPulsarConsumerErrorHandler<>(pulsarDeadLetterPublishingRecoverer, - new FixedBackOff(100, 5)); -} ----- - -Note that we are provide a destination resolver to the `PulsarDeadLetterPublishingRecoverer` as the second constructor argument. -If not provided, `PulsarDeadLetterPublishingRecoverer` uses `--DLT>` as the DLT topic name. -When using this feature, you should use a proper destination name by setting the destination resolver rather than using the default. - -When using a single record message listener, as we did with `PulsarConsumerErrorHnadler`, and if you use manual acknowledgement, make sure to not negatively acknowledge the message when an exception is thrown. -Rather, re-throw the exception back to the container. Otherwise, the container thinks the message is handled separately, and the error handling is not triggered. - -Finally, we have a second `PulsarListener` that receives messages from the DLT topic. - -In the examples provided in this section so far, we only saw how to use `PulsarConsumerErrorHandler` with a single record message listener. -Next, we look at how you can use this on batch listeners. - -==== Batch listener with PulsarConsumerErrorHandler - -First, let us look at a batch `PulsarListener` method: - -[source, java] ----- -@PulsarListener(subscriptionName = "batch-demo-5-sub", topics = "batch-demo-4", batch = true, concurrency = "3", - subscriptionType = SubscriptionType.Failover, - pulsarConsumerErrorHandler = "pulsarConsumerErrorHandler", ackMode = AckMode.MANUAL) -void listen(List> data, Consumer consumer, Acknowledgment acknowledgment) { - for (Message datum : data) { - if (datum.getValue() == 5) { - throw new PulsarBatchListenerFailedException("failed", datum); - } - acknowledgement.acknowledge(datum.getMessageId()); - } -} - -@Bean -PulsarConsumerErrorHandler pulsarConsumerErrorHandler( - PulsarTemplate pulsarTemplate) { - return new DefaultPulsarConsumerErrorHandler<>( - new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, (c, m) -> "my-foo-dlt"), new FixedBackOff(100, 10)); -} - -@PulsarListener(subscriptionName = "my-dlt-subscription", topics = "my-foo-dlt") -void dltReceiver(Message message) { - System.out.println("DLT - RECEIVED: " + message.getValue()); -} - ----- - -Once again, we provide the `pulsarConsumerErrorHandler` property with the `PulsarConsumerErrorHandler` bean name. -When you use a batch listener (as shown in the preceding example) and want to use the `PulsarConsumerErrorHandler` from Spring for Apache Pulsar, you need to use manual acknowledgment. -This way, you can acknowledge all the successful individual messages. -For the ones that fail, you must throw a `PulsarBatchListenerFailedException` with the message on which it fails. -Without this exception, the framework does not know what to do with the failure. -On retry, the container sends a new batch of messages, starting with the failed message to the listener. -If it fails again, it is retried, until the retries are exhausted, at which point the message is sent to the DLT. -At that point, the message is acknowledged by the container, and the listener is handed over with the subsequent messages in the original batch. - -=== Consumer Customization on PulsarListener - -Spring for Apache Pulsar provides a convenient way to customize the consumer created by the container used by the `PulsarListener`. -Applications can provide a bean for `PulsarListenerConsumerBuilderCustomizer`. -Here is an example. -[source, java] ----- -@Bean -public PulsarListenerConsumerBuilderCustomizer myCustomizer() { - return cb -> { - cb.subscriptionName("modified-subscription-name"); - }; -} ----- - -Then this customizer bean name can be provided as an attribute on the `PuslarListener` annotation as shown below. - -[source, java] ----- -@PulsarListener(subscriptionName = "my-subscription", - topics = "my-topic", consumerCustomizer = "myCustomizer") -void listen(String message) { - -} ----- - -The framework detects the provided bean through the `PulsarListener` and applies this customizer on the Consumer builder before creating the Pulsar Consumer. - -If you have multiple `PulsarListener` methods, and each of them have different customization rules, you should create multiple customizer beans and attach the proper customizers on each `PulsarListener`. - - -=== Pausing and Resuming Message Listener Containers - -There are situations in which an application might want to pause message consumption temporarily and then resume later. -Spring for Apache Pulsar provides the ability to pause and resume the underlying message listener containers. -When the Pulsar message listener container is paused, any polling done by the container to receive data from the Pulsar consumer will be paused. -Similarly, when the container is resumed, the next poll starts returning data if the topic has any new records added while paused. - -To pause or resume a listener container, first obtain the container instance via the `PulsarListenerEndpointRegistry` bean and then invoke the pause/resume API on the container instance - as shown in the snippet below: -[source, java] ----- -@Autowired -private PulsarListenerEndpointRegistry registry; - -void someMethod() { - PulsarMessageListenerContainer container = registry.getListenerContainer("my-listener-id"); - container.pause(); -} ----- - -TIP: The id parameter passed to `getListenerContainer` is the container id - which will be the value of the `@PulsarListener` id attribute when pausing/resuming a `@PulsarListener`. - -[[imperative-pulsar-reader]] -=== Pulsar Reader Support -The framework provides support for using {apache-pulsar-docs}/concepts-clients/#reader-interface[Pulsar Reader] via the `PulsarReaderFactory`. - -Spring Boot provides this reader factory which you can further configure by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.reader.*`] application properties. - -==== PulsarReader Annotation - -While it is possible to use `PulsarReaderFactory` directly, Spring for Apache Pulsar provides the `PulsarReader` annotation that you can use to quickly read from a topic without setting up any reader factories yourselves. -This is similar to the same ideas behind `PulsarListener.` -Here is a quick example. - -[source, java] ----- -@PulsarReader(id = "reader-demo-id", topics = "reader-demo-topic", startMessageId = "earliest") -void read(String message) { - //... -} ----- -The `id` attribute is optional, but it is a best practice to provide a value that is meaningful to your application. -When not specified an auto-generated id will be used. -On the other hand, the `topics` and `startMessageId` attributes are mandatory. -The `topics` attribute can be a single topic or a comma-separated list of topics. -The `startMessageId` attribute instructs the reader to start from a particular message in the topic. -The valid values for `startMessageId` are `earliest` or `latest.` -Suppose you want the reader to start reading messages arbitrarily from a topic other than the earliest or latest available messages. In that case, you need to use a `ReaderBuilderCustomizer` to customize the `ReaderBuilder` so it knows the right `MessageId` to start from. - -==== Customizing the ReaderBuilder - -You can customize any fields available through `ReaderBuilder` using a `PulsarReaderReaderBuilderCustomizer` in Spring for Apache Pulsar. -You can provide a `@Bean` of type `PulsarReaderReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below. - -[source, java] ----- -@PulsarReader(id = "reader-customizer-demo-id", topics = "reader-customizer-demo-topic", - readerCustomizer = "myCustomizer") -void read(String message) { - //... -} - -@Bean -public PulsarReaderReaderBuilderCustomizer myCustomizer() { - return readerBuilder -> { - readerBuilder.startMessageId(messageId); // the first message read is after this message id. - // Any other customizations on the readerBuilder - }; -} ----- - -TIP: If your application only has a single `@PulsarReader` and a single `PulsarReaderReaderBuilderCustomizer` bean registered then the customizer will be automatically applied. - -[[topic-resolution-process-imperative]] -== Topic Resolution -include::topic-resolution.adoc[leveloffset=+1] - -== Publishing and Consuming Partitioned Topics - -In the following example, we publish to a topic called `hello-pulsar-partitioned`. -It is a topic that is partitioned, and, for this sample, we assume that the topic is already created with three partitions. - -[source, java] ----- -@SpringBootApplication -public class PulsarBootPartitioned { - - public static void main(String[] args) { - SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.message-routing-mode=CustomPartition"); - } - - @Bean - public ApplicationRunner runner(PulsarTemplate 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; - } - } - -} ----- - -In the preceding example, we publish 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. -To do so, we provide a message router object with the `send` method. -Consider the three message routers implemented. -`FooRouter` always sends data to partition `0`, `BarRouter` sends to partition `1`, and `BuzzRouter` sends to partition `2`. -Also note that we now use 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.message-routing-mode`). - -On the consumer side, we use a `PulsarListener` with the exclusive subscription type. -This means that data from all the partitions ends 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: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = 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 = 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 = SubscriptionType.Failover) -public void listen3(String foo) { - System.out.println("Message Received 3: " + foo); -} ----- - -When you follow this approach, a single partition always gets consumed by a dedicated consumer. - -In a similar vein, if you want to use Pulsar's shared consumer type, you can use the `shared` subscription type. -However, when you use 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. - -Consider the following example: - -[source, java] ----- -@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = 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 = SubscriptionType.Shared) -public void listen2(String foo) { - System.out.println("Message Received 2: " + foo); -} ----- +include::pulsar/preface.adoc[leveloffset=+1] diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/authentication.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/authentication.adoc similarity index 89% rename from spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/authentication.adoc rename to spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/authentication.adoc index 7faf81fc..d74693cb 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/authentication.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/authentication.adoc @@ -1,5 +1,4 @@ - -include::../attributes/attributes.adoc[] +include::../../attributes/attributes.adoc[] To connect to a Pulsar cluster that requires authentication, you need to specify which authentication plugin to use and any parameters required by the specified plugin. When **using Spring Boot** auto-configuration, you can set the plugin and the plugin parameters via configuration properties (in most cases). @@ -59,7 +58,7 @@ spring: privateKey: ... keyId: ... ---- -NOTE: This also requires xref:reference/pulsar.adoc#tls-encryption[TLS encryption]. +NOTE: This also requires xref:reference/pulsar/pulsar-client.adoc#tls-encryption[TLS encryption]. ==== [[Token]] @@ -135,7 +134,7 @@ spring: .[.underline]#Click ##here## for **mTLS (PEM)**# [%collapsible] ==== -NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer. +NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar/pulsar-client.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer. You can use the `org.apache.pulsar.client.api.AuthenticationFactory` to help create the authentication object as follows: [source,java] ---- @@ -148,7 +147,7 @@ See the official Pulsar documentation on {apache-pulsar-docs}/security-tls-authe .[.underline]#Click ##here## for **mTLS (JKS)**# [%collapsible] ==== -NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer. +NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar/pulsar-client.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer. You can use the `org.apache.pulsar.client.api.AuthenticationFactory` to help create the authentication object as follows: [source,java] ---- diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc new file mode 100644 index 00000000..7b8c675e --- /dev/null +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-consumption.adoc @@ -0,0 +1,1031 @@ +[[message-consumption]] += Message Consumption +include::../../attributes/attributes.adoc[] + +[[pulsar-listener]] +== Pulsar Listener + +When it comes to Pulsar consumers, we recommend that end-user applications use the `PulsarListener` annotation. +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 the underlying Pulsar consumer that it uses to consume messages. + +Spring Boot provides this consumer factory which you can further configure by specifying the {spring-boot-pulsar-config-props}[`spring.pulsar.consumer.*`] application properties. +**Most** of the configured properties on the factory will be respected in the listener with the following **exceptions**: + +TIP: The `spring.pulsar.consumer.subscription.name` property is ignored and is instead generated when not specified on the annotation. + +TIP: The `spring.pulsar.consumer.subscription-type` property is ignored and is instead taken from the value on the annotation. However, you can set the `subscriptionType = {}` on the annotation to instead use the property value as the default. + + +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); +} +---- + +You can further simplify this method: + +[source, java] +---- +@PulsarListener +public void listen(String message) { + System.out.println("Message Received: " + message); +} +---- + +In this most basic form, when the `subscriptionName` is not provided on the `@PulsarListener` annotation an auto-generated subscription name will be used. +Likewise, when the `topics` are not directly provided, a xref:reference/pulsar/topic-resolution.adoc#topic-resolution-process-imperative[topic resolution process] is used to determine the destination topic. + +In the `PulsarListener` 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 and provides that schema to the consumer. +The framework does this inference for all primitive types. +For all non-primitive types the default schema is assumed to be JSON. +If a complex type is using anything besides JSON (such as AVRO or KEY_VALUE) you must provide the schema type on the annotation using the `schemaType` property. + +The following example shows another `PulsarListener` method, which takes an `Integer`: + +[source, java] +---- +@PulsarListener(subscriptionName = "my-subscription-1", topics = "my-topic-1") +public void listen(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 listen(Foo message) { + System.out.println(message); +} +---- + +Let us look at a few more ways. + +You can consume the Pulsar message directly: + +[source, java] +---- +@PulsarListener(subscriptionName = "my-subscription", topics = "my-topic") +public void listen(org.apache.pulsar.client.api.Message message) { + System.out.println(message.getValue()); +} +---- + +The following example consumes the record by using the Spring messaging envelope: + +[source, java] +---- +@PulsarListener(subscriptionName = "my-subscription", topics = "my-topic") +public void listen(org.springframework.messaging.Message message) { + System.out.println(message.getPayload()); +} +---- + +Now let us see how we can consume records in batches. +The following example uses `PulsarListener` to consume records in batches as POJOs: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) +public void listen(List messages) { + System.out.println("records received :" + messages.size()); + messages.forEach((message) -> System.out.println("record : " + message)); +} +---- + +Note that, in this example, we receive the records as a collection (`List`) of objects. +In addition, 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 besides JSON, you still need to provide the `schemaType` on `PulsarListener`. + +The following uses the `Message` envelope provided by the Pulsar Java client: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) +public void listen(List> messages) { + System.out.println("records received :" + messages.size()); + messages.forEach((message) -> System.out.println("record : " + message.getValue())); +} +---- + +The following example consumes batch records with an envelope of the Spring messaging `Message` type: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) +public void listen(List> messages) { + System.out.println("records received :" + messages.size()); + messages.forEach((message) -> System.out.println("record : " + message.getPayload())); +} +---- + +Finally, you can also use the `Messages` holder object from Pulsar for the batch listener: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) +public void listen(org.apache.pulsar.client.api.Messages> messages) { + System.out.println("records received :" + messages.size()); + messages.forEach((message) -> System.out.println("record : " + message.getValue())); +} +---- + +When you use `PulsarListener`, you can provide Pulsar consumer properties directly on the annotation itself. +This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple `PulsarListener` methods. + +The following example uses Pulsar consumer properties directly on `PulsarListener`: + +[source, java] +---- +@PulsarListener(properties = { "subscriptionName=subscription-1", "topicNames=foo-1", "receiverQueueSize=5000" }) +void listen(String message) { +} +---- + +TIP: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` application configuration properties + +[[listener-auto-consume]] +=== Generic records with AUTO_CONSUME +If there is no chance to know the type of schema of a Pulsar topic in advance, you can use the `AUTO_CONSUME` schema type to consume generic records. +In this case, the topic deserializes messages into `GenericRecord` objects using the schema info associated with the topic. + +To consume generic records set the `schemaType = SchemaType.AUTO_CONSUME` on your `@PulsarListener` and use a Pulsar message of type `GenericRecord` as the message parameter as shown below. + +[source, java] +---- +@PulsarListener(topics = "my-generic-topic", schemaType = SchemaType.AUTO_CONSUME) +void listen(org.apache.pulsar.client.api.Message message) { + GenericRecord record = message.getValue(); + record.getFields().forEach((f) -> + System.out.printf("%s = %s%n", f.getName(), record.getField(f))); +} +---- + +TIP: The `GenericRecord` API allows access to the fields and their associated values + +=== Customizing the ConsumerBuilder + +You can customize any fields available through `ConsumerBuilder` using a `PulsarListenerConsumerBuilderCustomizer` by providing a `@Bean` of type `PulsarListenerConsumerBuilderCustomizer` and then making it available to the `PulsarListener` as shown below. + +[source, java] +---- +@PulsarListener(topics = "hello-topic", consumerCustomizer = "myCustomizer") +public void listen(String message) { + System.out.println("Message Received: " + message); +} + +@Bean +PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return (builder) -> builder.consumerName("myConsumer"); +} +---- + +TIP: If your application only has a single `@PulsarListener` and a single `PulsarListenerConsumerBuilderCustomizer` bean registered then the customizer will be automatically applied. + + +[[schema-info-listener-imperative]] +:listener-class: PulsarListener +include::../schema-info/schema-info-listener.adoc[] + +== Accessing the Pulsar Consumer Object +Sometimes, you need direct access to the Pulsar Consumer object. +The following example shows how to get it: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") +public void listen(String message, org.apache.pulsar.client.api.Consumer consumer) { + System.out.println("Message Received: " + message); + ConsumerStats stats = consumer.getStats(); + ... +} +---- + +CAUTION: When accessing the `Consumer` object this way, do NOT 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. + +[[pulsar-message-listener-container]] +== Pulsar Message Listener Container + +Now that we saw the basic interactions on the consumer side through `PulsarListener`. Let us now dive into the inner workings of how `PulsarListener` interacts with the underlying Pulsar consumer. +Keep in mind that, for end-user applications, in most scenarios, we recommend using the `PulsarListener` annotation directly for consuming from a Pulsar topic when using Spring for Apache Pulsar, as that model covers a broad set of application use cases. +However, it is important to understand how `PulsarListener` works internally. This section goes through those details. + +As briefly mentioned earlier, the message listener container is at the heart of message consumption when you use 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 by 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 you use Spring for Apache Pulsar. + +* link:{github}/blob/8e33ac0b122bc0e75df299919c956cacabcc9809/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java#L29[PulsarRecordMessageListener] + +* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarAcknowledgingMessageListener.java#L28[PulsarAcknowledgingMessageListener] + +* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java#L36[PulsarBatchMessageListener] + +* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java#L28[PulsarBatchAcknowledgingMessageListener] + +We see the details about these various message listeners in the following sections. + +Before doing so, however, let us take a closer look at the container itself. + +=== DefaultPulsarMessageListenerContainer + +This is a single consumer-based message listener container. +The following listing shows its constructor: + +[source, java] +---- +public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, + PulsarContainerProperties pulsarContainerProperties) +} +---- + +It receives a `PulsarConsumerFactory` (which it uses to create the consumer) and a `PulsarContainerProperties` object (which contains information about the container properties). +`PulsarContainerProperties` has the following constructors: + +[source, java] +---- +public PulsarContainerProperties(String... topics) + +public PulsarContainerProperties(Pattern topicPattern) +---- + +You can provide the topic information through `PulsarContainerProperties` or as a consumer property that is provided to the consumer factory. +The following example uses the `DefaultPulsarMessageListenerContainer`: + +[source, java] +---- +Map config = new HashMap<>(); +config.put("topics", "my-topic"); +PulsarConsumerFactory pulsarConsumerFactorY = DefaultPulsarConsumerFactory<>(pulsarClient, config); + +PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + +pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { + }); + +DefaultPulsarMessageListenerContainer pulsarListenerContainer = new DefaultPulsarMessageListenerContainer(pulsarConsumerFacotyr, + pulsarContainerProperties); + +return pulsarListenerContainer; +---- + +NOTE: If topic information is not specified when using the listener containers directly, the same +xref:reference/pulsar/topic-resolution.adoc#topic-resolution-process-imperative[topic resolution process] used by the `PulsarListener` is used with the one exception that the "Message type default" step is **omitted**. + +`DefaultPulsarMessageListenerContainer` creates only a single consumer. +If you want to have multiple consumers managed through multiple threads, you need to use `ConcurrentPulsarMessageListenerContainer`. + +=== ConcurrentPulsarMessageListenerContainer + +`ConcurrentPulsarMessageListenerContainer` has the following constructor: + +[source, java] +---- +public ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, + PulsarContainerProperties pulsarContainerProperties) +---- + +`ConcurrentPulsarMessageListenerContainer` lets you specify a `concurrency` property through a setter. +Concurrency of more than `1` is allowed only on non-exclusive subscriptions (`failover`, `shared`, and `key-shared`). +You can only have the default `1` for concurrency when you have an exclusive subscription mode. + +The following example enables `concurrency` through the `PulsarListener` annotation for a `failover` subscription. + +[source, java] +---- +@PulsarListener(topics = "my-topic", subscriptionName = "subscription-1", + subscriptionType = SubscriptionType.Failover, concurrency = "3") +void listen(String message, Consumer consumer) { + ... + System.out.println("Current Thread: " + Thread.currentThread().getName()); + System.out.println("Current Consumer: " + consumer.getConsumerName()); +} +---- + +In the preceding listener, it is assumed that the topic `my-topic` has three partitions. +If it is a non-partitioned topic, having concurrency set to `3` does nothing. You get two idle consumers in addition to the main active one. +If the topic has more than three partitions, messages are load-balanced across the consumers that the container creates. +If you run this `PulsarListener`, you see that messages from different partitions are consumed through different consumers, as implied by the thread name and consumer names printouts in the preceding example. + +NOTE: When you use the `Failover` subscription this way on partitioned topics, Pulsar guarantees message ordering. + +The following listing shows another example of `PulsarListener`, but with `Shared` subscription and `concurrency` enabled. + +[source, java] +---- +@PulsarListener(topics = "my-topic", subscriptionName = "subscription-1", + subscriptionType = SubscriptionType.Shared, concurrency = "5") +void listen(String message) { + ... +} +---- + +In the preceding example, the `PulsarListener` creates five different consumers (this time, we assume that the topic has five partitions). + +NOTE: In this version, there is no message ordering, as `Shared` subscriptions do not guarantee any message ordering in Pulsar. + +If you need message ordering and still want a shared subscription types, you need to use the `Key_Shared` subscription type. + +[[consuming-records]] +=== Consuming Records + +Let us take a look at how the message listener container enables both single-record and batch-based message consumption. + +[discrete] +=== Single Record Consumption +Let us revisit 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, we essential ask 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. This 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 invokes the listener method through an adapter for `PulsarRecordMessageListener`. +As you can see in the previous section, `PulsarRecordMessageListener` extends from the `MessageListener` provided by the Pulsar Java client, and it supports the basic `received` method. + +[discrete] +=== Batch Consumption +The following example shows the `PulsarListener` consuming records in batches: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) +public void listen4(List messages) { + System.out.println("records received :" + messages.size()); + messages.forEach((message) -> System.out.println("record : " + message)); +} +---- + +When you use this type of `PulsarListener`, the framework detects that you are in batch mode. +Since it already received the data in batches by using the Consumer's `batchReceive` method, it hands off the entire batch to the listener method through an adapter for `PulsarBatchMessageListener`. + +[[pulsar-headers]] +== Pulsar Headers +The Pulsar message metadata can be consumed as Spring message headers. +The list of available headers can be found in {github}/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java[PulsarHeaders.java]. + +=== Accessing in Single Record based Consumer + +The following example shows how you can access the various Pulsar Headers in an application that uses the single record mode of consuming: + +[source,java] +---- +@PulsarListener(topics = "simpleListenerWithHeaders") +void simpleListenerWithHeaders(String data, @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.RAW_DATA) byte[] rawData, + @Header("foo") String foo) { + +} +---- + +In the preceding example, we access the values for the `messageId` and `rawData` 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 Batch Record based Consumer + +In this section, we see how to access the various Pulsar Headers in an application that uses a batch consumer: + +[source,java] +---- +@PulsarListener(topics = "simpleBatchListenerWithHeaders", batch = true) +void simpleBatchListenerWithHeaders(List data, + @Header(PulsarHeaders.MESSAGE_ID) List messageIds, + @Header(PulsarHeaders.TOPIC_NAME) List topicNames, @Header("foo") List fooValues) { + +} +---- + +In the preceding example, we consume the data as a `List`. +When extracting the various headers, we do so as a `List<>` as well. +Spring for Apache 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.Messages`, or `org.springframework.messaging.Messsge`. + +== Message Acknowledgment + +When you use 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]] +=== 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. +In the following sections, we see how acknowledgment works when you use both single and batch versions of `PulsarListener` and how they translate to the backing message listener container (and, ultimately, to the Pulsar consumer). + +=== Automatic Message Ack in Single Record Mode + +Let us revisit our basic single message based `PulsarListener`: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") +public void listen(String message) { + System.out.println("Message Received: " + message); +} +---- + +It is natural to wonder, how acknowledgment works when you use `PulsarListener`, especially if you are familiar with using Pulsar consumer directly. +The answer comes down to the message listener container, as that is the central place in Spring for Apache Pulsar that coordinates all the consumer related activities. + +Assuming you are not overriding the default behavior, this is what happens behind the scenes when you use the preceding `PulsarListener`: + +. First, the listener container receives messages as batches from the Pulsar consumer. +. The received messages are handed down to `PulsarListener` one message at a time. +. When all the records are handed down to the listener method and successfully processed, the container acknowledges all the messages from the original batch. + +This is the normal flow. If any records from the original batch throw an exception, Spring for Apache Pulsar track those records separately. +When all the records from the batch are processed, Spring for Apache Pulsar acknowledges all the successful messages and negatively acknowledges (nack) all the failed messages. +In other words, when consuming single records by using `PulsarRecordMessageListener` and the default ack mode of `BATCH` is used, the framework waits for all the records received from the `batchReceive` call to process successfully and then calls 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 calls `negativeAcknowledge` on those records after the entire batch is processed. + +If the application wants the acknowledgment or negative acknowledgment to occur per record, the `RECORD` ack mode can be enabled. +In that case, after handling each record, the message is acknowledged if no error and negatively acknowledged if there was an error. +The following example enables `RECORD` ack mode on the Pulsar listener: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", ackMode = AckMode.RECORD) +public void listen(String message) { + System.out.println("Message Received: " + message); +} +---- + +=== Manual Message Ack in Single Record Mode + +You might not always want the framework to send acknowledgments but, rather, do that directly from the application itself. +Spring for Apache Pulsar provides a couple of ways to enable manual message acknowledgments. The following example shows one of them: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", ackMode = AckMode.MANUAL) +public void listen(Message message, Acknowledgment acknowledgment) { + System.out.println("Message Received: " + message.getValue()); + acknowledgment.acknowledge(); +} +---- + +A few things merit explanation here. First, we enablE manual ack mode by setting `ackMode` on `PulsarListener`. +When enabling manual ack mode, Spring for Apache Pulsar lets the application inject an `Acknowledgment` object. +The framework achieves this by selecting a compatible message listener container: `PulsarAcknowledgingMessageListener` for single record based consumption, which gives 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 messageIds); + +void nack(); + +void nack(MessageId messageId); +---- + +You can inject this `Acknowledgment` object into your `PulsarListener` while using `MANUAL` ack mode and then call one of the corresponding methods. + +In the preceding `PulsarListener` example, we call a parameter-less `acknowledge` method. +This is because the framework knows which `Message` it is currently operating under. +When calling `acknowledge()`, you need not receive the payload with the `Message` enveloper` but, rather, use the target type -- `String`, in this example. +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. + +You can also call `acknowledge` directly on the Pulsar consumer: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", ackMode = AckMode.MANUAL) +public void listen(Message message, Consumer consumer) { + System.out.println("Message Received: " + message.getValue()); + try { + consumer.acknowledge(message); + } + catch (Exception e) { + .... + } +} +---- + +When calling `acknowledge` directly on the underlying consumer, you need to do error handling by yourself. +Using the `Acknowledgment` does not require that, as the framework can do that for you. +Therefore, you should use the `Acknowledgment` object approach when using manual acknowledgment. + +IMPORTANT: When using manual acknowledgment, it is important to understand that the framework completely stays from any acknowledgment at all. +Hence, it is extremely important to think through the right acknowledgment strategies when designing applications. + +=== Automatic Message Ack in Batch Consumption + +When you consume records in batches (see "`<>`") and you use the default ack mode of `BATCH` is used, when the entire batch is processed successfully, the entire batch is acknowledged. +If any records throw an exception, the entire batch is negatively acknowledged. +Note that this may not be the same batch that was batched on the producer side. Rather, this is the batch that returned from calling `batchReceive` on the consumer + +Consider the following batch listener: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar", batch = true) +public void batchListen(List messages) { + for (Foo foo : messages) { + ... + } +} +---- + +When all the messages in the incoming collection (`messages` in this example) are processed, the framework acknowledges all of them. + +When consuming in batch mode, `RECORD` is not an allowed ack mode. +This might cause an issue, as an application may not want the entire batch to be re-delivered again. +In such situations, you need to use the `MANUAL` acknowledgement mode. + +=== Manual Message Ack in Batch Consumption + +As seen in the previous section, when `MANUAL` ack mode is set on the message listener container, the framework does 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: `PulsarBatchAcknowledgingMessageListener` for batch consumption, which gives you access to an `Acknowledgment` object. +The following are the methods available in the `Acknowledgment` API: + +[source, java] +---- +void acknowledge(); + +void acknowledge(MessageId messageId); + +void acknowledge(List messageIds); + +void nack(); + +void nack(MessageId messageId); +---- + +You can inject this `Acknowledgment` object into your `PulsarListener` while using `MANUAL` ack mode. +The following listing shows a basic example for a batch based listener: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") +public void listen(List> messgaes, Acknowlegement acknowledgment) { + for (Message message : messages) { + try { + ... + acknowledgment.acknowledge(message.getMessageId()); + } + catch (Exception e) { + acknowledgment.nack(message.getMessageId()); + } + } +} +---- + +When you use a batch listener, the message listener container cannot know which record it is currently operating upon. +Therefore, to manually acknowledge, you need to use one of the overloaded `acknowledge` method that takes a `MessageId` or a `List`. +You can also negatively acknowledge with the `MessageId` for the batch listener. + +== Message Redelivery and Error Handling + +Now that we have seen both `PulsarListener` and the message listener container infrastructure and its various functions, let us now try to understand message redelivery and error handling. +Apache Pulsar provides various native strategies for message redelivery and error handling. We take a look at them and see how we can use them through Spring for Apache Pulsar. + +=== Specifying Acknowledgment Timeout for Message Redelivery + +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. +If the ack timeout property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered. + +When you use Spring for Apache Pulsar, you can set this property via a <<_consumer_customization_on_pulsarlistener,consumer customizer>> or with the native Pulsar `ackTimeout` property in the `properties` attribute of `@PulsarListener`: + +[source, java] +---- +@PulsarListener(subscriptionName = "subscription-1", topics = "topic-1" + properties = {"ackTimeout=60s"}) +public void listen(String s) { + ... +} +---- + +When you specify the ack timeout, if the consumer does not send an acknowledgement within 60 seconds, the message is redelivered by Pulsar to the consumer. + +If you want to specify some advanced backoff options for ack timeout with different delays, you can do the following: + +[source, java] +---- +@EnablePulsar +@Configuration +class AckTimeoutRedeliveryConfig { + + @PulsarListener(subscriptionName = "withAckTimeoutRedeliveryBackoffSubscription", + topics = "withAckTimeoutRedeliveryBackoff-test-topic", + ackTimeoutRedeliveryBackoff = "ackTimeoutRedeliveryBackoff", + properties = { "ackTimeout=60s" }) + void listen(String msg) { + // some long-running process that may cause an ack timeout + } + + @Bean + RedeliveryBackoff ackTimeoutRedeliveryBackoff() { + return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(10 * 1000).multiplier(2) + .build(); + } + +} +---- + +In the preceding example, we specify a bean for Pulsar's `RedeliveryBackoff` with a minimum delay of 1 second, a maximum delay of 10 seconds, and a backoff multiplier of 2. +After the initial ack timeout occurs, the message redeliveries are controlled through this backoff bean. +We provide the backoff bean to the `PulsarListener` annotation by setting the `ackTimeoutRedeliveryBackoff` property to the actual bean name -- `ackTimeoutRedeliveryBackoff`, in this case. + +=== Specifying Negative Acknowledgment Redelivery + +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 via a <<_consumer_customization_on_pulsarlistener,consumer customizer>> or with the native Pulsar `negativeAckRedeliveryDelay` property in the `properties` attribute of `@PulsarListener`: + +[source, java] +---- +@PulsarListener(subscriptionName = "subscription-1", topics = "topic-1" + properties = {"negativeAckRedeliveryDelay=10ms"}) +public void listen(String s) { + ... +} +---- + +You can also specify different delays and backoff mechanisms with a multiplier by providing a `RedeliveryBackoff` bean and providing the bean name as the `negativeAckRedeliveryBackoff` property on the PulsarProducer, as follows: + +[source, java] +---- +@EnablePulsar +@Configuration +class NegativeAckRedeliveryConfig { + + @PulsarListener(subscriptionName = "withNegRedeliveryBackoffSubscription", + topics = "withNegRedeliveryBackoff-test-topic", negativeAckRedeliveryBackoff = "redeliveryBackoff", + subscriptionType = SubscriptionType.Shared) + void listen(String msg) { + throw new RuntimeException("fail " + msg); + } + + @Bean + RedeliveryBackoff redeliveryBackoff() { + return MultiplierRedeliveryBackoff.builder().minDelayMs(1000).maxDelayMs(10 * 1000).multiplier(2) + .build(); + } + +} +---- + +=== Using Dead Letter Topic from Apache Pulsar for Message Redelivery and Error Handling + +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] +---- +@EnablePulsar +@Configuration +class DeadLetterPolicyConfig { + + @PulsarListener(id = "deadLetterPolicyListener", subscriptionName = "deadLetterPolicySubscription", + topics = "topic-with-dlp", deadLetterPolicy = "deadLetterPolicy", + subscriptionType = SubscriptionType.Shared, properties = { "ackTimeout=1s" }) + void listen(String msg) { + throw new RuntimeException("fail " + msg); + } + + @PulsarListener(id = "dlqListener", topics = "my-dlq-topic") + void listenDlq(String msg) { + System.out.println("From DLQ: " + msg); + } + + @Bean + DeadLetterPolicy deadLetterPolicy() { + return DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic("my-dlq-topic").build(); + } + +} +---- + +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 `--DLQ` in Pulsar. +Next, we provide this bean name to `PulsarListener` by setting the `deadLetterPolicy` property. +Note that the `PulsarListener` 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 DLQ topic. +We have another `PulsarListener` 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-`, 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-` 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. +**** + +=== Native Error Handling in Spring for Apache Pulsar + +As we noted earlier, the DLQ feature in Apache Pulsar works only for shared subscriptions. +What does an application do if it needs to use some similar feature for non-shared subscriptions? +The main reason Pulsar does not support DLQ on exclusive and failover subscriptions is because those subscription types are order-guaranteed. +Allowing redeliveries, DLQ, and so on effectively receives messages out of order. +However, what if an application are okay with that but, more importantly, needs this DLQ feature for non-shared subscriptions? +For that, Spring for Apache Pulsar provides a `PulsarConsumerErrorHandler`, which you can use across any subscription types in Pulsar: `Exclusive`, `Failover`, `Shared`, or `Key_Shared`. + +When you use `PulsarConsumerErrorHandler` from Spring for Apache Pulsar, make sure not to set the ack timeout properties on the listener. + +Let us see some details by examining a few code snippets: + +[source, java] +---- +@EnablePulsar +@Configuration +class PulsarConsumerErrorHandlerConfig { + + @Bean + PulsarConsumerErrorHandler pulsarConsumerErrorHandler( + PulsarTemplate pulsarTemplate) { + return new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, (c, m) -> "my-foo-dlt"), new FixedBackOff(100, 10)); + } + + @PulsarListener(id = "pulsarConsumerErrorHandler-id", subscriptionName = "pulsatConsumerErrorHandler-subscription", + topics = "pulsarConsumerErrorHandler-topic", + pulsarConsumerErrorHandler = "pulsarConsumerErrorHandler") + void listen(String msg) { + throw new RuntimeException("fail " + msg); + } + + @PulsarListener(id = "pceh-dltListener", topics = "my-foo-dlt") + void listenDlt(String msg) { + System.out.println("From DLT: " + msg); + } + +} +---- + +Consider the `pulsarConsumerErrorHandler` bean. +This creates a bean of type `PulsarConsumerErrorHandler` and uses the default implementation provided out of the box by Spring for Apache Pulsar: `DefaultPulsarConsumerErrorHandler`. +`DefaultPulsarConsumerErrorHandler` has a constructor that takes a `PulsarMessageRecovererFactory` and a `org.springframework.util.backoff.Backoff`. +`PulsarMessageRecovererFactory` is a functional interface with the following API: + +[source, java] +---- +@FunctionalInterface +public interface PulsarMessageRecovererFactory { + + /** + * Provides a message recoverer {@link PulsarMessageRecoverer}. + * @param consumer Pulsar consumer + * @return {@link PulsarMessageRecoverer}. + */ + PulsarMessageRecoverer recovererForConsumer(Consumer consumer); + +} + +---- + +The `recovererForConsumer` method takes a Pulsar consumer and returns a `PulsarMessageRecoverer`, which is another functional interface. +Here is the API of `PulsarMessageRecoverer`: + +[source, java] +---- +public interface PulsarMessageRecoverer { + + /** + * Recover a failed message, for e.g. send the message to a DLT. + * @param message Pulsar message + * @param exception exception from failed message + */ + void recoverMessage(Message message, Exception exception); + +} +---- + +Spring for Apache Pulsar provides an implementation for `PulsarMessageRecovererFactory` called `PulsarDeadLetterPublishingRecoverer` that provides a default implementation that can recover the message by sending it to a Dead Letter Topic (DLT). +We provide this implementation to the constructor for the preceding `DefaultPulsarConsumerErrorHandler`. +As the second argument, we provide a `FixedBackOff`. +You can also provide the `ExponentialBackoff` from Spring for advanced backoff features. +Then we provide this bean name for the `PulsarConsumerErrorHandler` as a property to the `PulsarListener`. +The property is called `pulsarConsumerErrorHandler`. +Each time the `PulsarListener` method fails for a message, it gets retried. +The number of retries are controlled by the `Backoff` provided implementation values. In our example, we do 10 retries (11 total tries -- the first one and then the 10 retries). +Once all the retries are exhausted, the message is sent to the DLT topic. + +The `PulsarDeadLetterPublishingRecoverer` implementation we provide uses a `PulsarTemplate` that is used for publishing the message to the DLT. +In most cases, the same auto-configured `PulsarTemplate` from Spring Boot is sufficient with the caveat for partitioned topics. +When using partitioned topics and using custom message routing for the main topic, you must use a different `PulsarTemplate` that does not take the auto-configured `PulsarProducerFactory` that is populated with a value of `custompartition` for `message-routing-mode`. +You can use a `PulsarConsumerErrorHandler` with the following blueprint: + +[source, java] +---- +@Bean +PulsarConsumerErrorHandler pulsarConsumerErrorHandler(PulsarClient pulsarClient) { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Map.of()); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + + BiFunction, Message, String> destinationResolver = + (c, m) -> "my-foo-dlt"; + + PulsarDeadLetterPublishingRecoverer pulsarDeadLetterPublishingRecoverer = + new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, destinationResolver); + + return new DefaultPulsarConsumerErrorHandler<>(pulsarDeadLetterPublishingRecoverer, + new FixedBackOff(100, 5)); +} +---- + +Note that we are provide a destination resolver to the `PulsarDeadLetterPublishingRecoverer` as the second constructor argument. +If not provided, `PulsarDeadLetterPublishingRecoverer` uses `--DLT>` as the DLT topic name. +When using this feature, you should use a proper destination name by setting the destination resolver rather than using the default. + +When using a single record message listener, as we did with `PulsarConsumerErrorHnadler`, and if you use manual acknowledgement, make sure to not negatively acknowledge the message when an exception is thrown. +Rather, re-throw the exception back to the container. Otherwise, the container thinks the message is handled separately, and the error handling is not triggered. + +Finally, we have a second `PulsarListener` that receives messages from the DLT topic. + +In the examples provided in this section so far, we only saw how to use `PulsarConsumerErrorHandler` with a single record message listener. +Next, we look at how you can use this on batch listeners. + +=== Batch listener with PulsarConsumerErrorHandler + +First, let us look at a batch `PulsarListener` method: + +[source, java] +---- +@PulsarListener(subscriptionName = "batch-demo-5-sub", topics = "batch-demo-4", batch = true, concurrency = "3", + subscriptionType = SubscriptionType.Failover, + pulsarConsumerErrorHandler = "pulsarConsumerErrorHandler", ackMode = AckMode.MANUAL) +void listen(List> data, Consumer consumer, Acknowledgment acknowledgment) { + for (Message datum : data) { + if (datum.getValue() == 5) { + throw new PulsarBatchListenerFailedException("failed", datum); + } + acknowledgement.acknowledge(datum.getMessageId()); + } +} + +@Bean +PulsarConsumerErrorHandler pulsarConsumerErrorHandler( + PulsarTemplate pulsarTemplate) { + return new DefaultPulsarConsumerErrorHandler<>( + new PulsarDeadLetterPublishingRecoverer<>(pulsarTemplate, (c, m) -> "my-foo-dlt"), new FixedBackOff(100, 10)); +} + +@PulsarListener(subscriptionName = "my-dlt-subscription", topics = "my-foo-dlt") +void dltReceiver(Message message) { + System.out.println("DLT - RECEIVED: " + message.getValue()); +} + +---- + +Once again, we provide the `pulsarConsumerErrorHandler` property with the `PulsarConsumerErrorHandler` bean name. +When you use a batch listener (as shown in the preceding example) and want to use the `PulsarConsumerErrorHandler` from Spring for Apache Pulsar, you need to use manual acknowledgment. +This way, you can acknowledge all the successful individual messages. +For the ones that fail, you must throw a `PulsarBatchListenerFailedException` with the message on which it fails. +Without this exception, the framework does not know what to do with the failure. +On retry, the container sends a new batch of messages, starting with the failed message to the listener. +If it fails again, it is retried, until the retries are exhausted, at which point the message is sent to the DLT. +At that point, the message is acknowledged by the container, and the listener is handed over with the subsequent messages in the original batch. + +== Consumer Customization on PulsarListener + +Spring for Apache Pulsar provides a convenient way to customize the consumer created by the container used by the `PulsarListener`. +Applications can provide a bean for `PulsarListenerConsumerBuilderCustomizer`. +Here is an example. +[source, java] +---- +@Bean +public PulsarListenerConsumerBuilderCustomizer myCustomizer() { + return cb -> { + cb.subscriptionName("modified-subscription-name"); + }; +} +---- + +Then this customizer bean name can be provided as an attribute on the `PuslarListener` annotation as shown below. + +[source, java] +---- +@PulsarListener(subscriptionName = "my-subscription", + topics = "my-topic", consumerCustomizer = "myCustomizer") +void listen(String message) { + +} +---- + +The framework detects the provided bean through the `PulsarListener` and applies this customizer on the Consumer builder before creating the Pulsar Consumer. + +If you have multiple `PulsarListener` methods, and each of them have different customization rules, you should create multiple customizer beans and attach the proper customizers on each `PulsarListener`. + + +== Pausing and Resuming Message Listener Containers + +There are situations in which an application might want to pause message consumption temporarily and then resume later. +Spring for Apache Pulsar provides the ability to pause and resume the underlying message listener containers. +When the Pulsar message listener container is paused, any polling done by the container to receive data from the Pulsar consumer will be paused. +Similarly, when the container is resumed, the next poll starts returning data if the topic has any new records added while paused. + +To pause or resume a listener container, first obtain the container instance via the `PulsarListenerEndpointRegistry` bean and then invoke the pause/resume API on the container instance - as shown in the snippet below: +[source, java] +---- +@Autowired +private PulsarListenerEndpointRegistry registry; + +void someMethod() { + PulsarMessageListenerContainer container = registry.getListenerContainer("my-listener-id"); + container.pause(); +} +---- + +TIP: The id parameter passed to `getListenerContainer` is the container id - which will be the value of the `@PulsarListener` id attribute when pausing/resuming a `@PulsarListener`. + +[[imperative-pulsar-reader]] +== Pulsar Reader Support +The framework provides support for using {apache-pulsar-docs}/concepts-clients/#reader-interface[Pulsar Reader] via the `PulsarReaderFactory`. + +Spring Boot provides this reader factory which you can further configure by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.reader.*`] application properties. + +=== PulsarReader Annotation + +While it is possible to use `PulsarReaderFactory` directly, Spring for Apache Pulsar provides the `PulsarReader` annotation that you can use to quickly read from a topic without setting up any reader factories yourselves. +This is similar to the same ideas behind `PulsarListener.` +Here is a quick example. + +[source, java] +---- +@PulsarReader(id = "reader-demo-id", topics = "reader-demo-topic", startMessageId = "earliest") +void read(String message) { + //... +} +---- +The `id` attribute is optional, but it is a best practice to provide a value that is meaningful to your application. +When not specified an auto-generated id will be used. +On the other hand, the `topics` and `startMessageId` attributes are mandatory. +The `topics` attribute can be a single topic or a comma-separated list of topics. +The `startMessageId` attribute instructs the reader to start from a particular message in the topic. +The valid values for `startMessageId` are `earliest` or `latest.` +Suppose you want the reader to start reading messages arbitrarily from a topic other than the earliest or latest available messages. In that case, you need to use a `ReaderBuilderCustomizer` to customize the `ReaderBuilder` so it knows the right `MessageId` to start from. + +=== Customizing the ReaderBuilder + +You can customize any fields available through `ReaderBuilder` using a `PulsarReaderReaderBuilderCustomizer` in Spring for Apache Pulsar. +You can provide a `@Bean` of type `PulsarReaderReaderBuilderCustomizer` and then make it available to the `PulsarReader` as below. + +[source, java] +---- +@PulsarReader(id = "reader-customizer-demo-id", topics = "reader-customizer-demo-topic", + readerCustomizer = "myCustomizer") +void read(String message) { + //... +} + +@Bean +public PulsarReaderReaderBuilderCustomizer myCustomizer() { + return readerBuilder -> { + readerBuilder.startMessageId(messageId); // the first message read is after this message id. + // Any other customizations on the readerBuilder + }; +} +---- + +TIP: If your application only has a single `@PulsarReader` and a single `PulsarReaderReaderBuilderCustomizer` bean registered then the customizer will be automatically applied. diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-production.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-production.adoc new file mode 100644 index 00000000..248c8cf0 --- /dev/null +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-production.adoc @@ -0,0 +1,182 @@ +[[message-production]] += Message Production +include::../../attributes/attributes.adoc[] + +[[pulsar-producer]] +== Pulsar Template + +On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides methods to publish records through its contract. + +There are two categories of these send API methods: `send` and `sendAsync`. +The `send` methods block calls by 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`, which you can use to asynchronously receive the message ID once the messages are published. + +NOTE: For the API variants that do not include a topic parameter, a xref:reference/pulsar/topic-resolution.adoc#topic-resolution-process-imperative[topic resolution process] is used to determine the destination topic. + +=== Simple API +The template provides a handful of methods ({javadocs}/org/springframework/pulsar/core/PulsarOperations.html[prefixed with _'send'_]) for simple send requests. For more complicated send requests, a fluent API lets you configure more options. + +=== Fluent API +The template provides a {javadocs}/org/springframework/pulsar/core/PulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests. + +=== Message customization +You can specify a `TypedMessageBuilderCustomizer` to configure the outgoing message. For example, the following code shows how to send a keyed message: +[source, java] +---- +template.newMessage(msg) + .withMessageCustomizer((mb) -> mb.key("foo-msg-key")) + .send(); +---- + +[[single-producer-customize]] +=== Producer customization +You can specify a `ProducerBuilderCustomizer` to configure the underlying Pulsar producer builder that ultimately constructs the producer used to send the outgoing message. + +WARNING: Use with caution as this gives full access to the producer 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) + .withProducerCustomizer((pb) -> pb.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 `Producer` builder such as: +[source, java] +---- +template.newMessage(msg) + .withProducerCustomizer((pb) -> pb.messageRouter(messageRouter)) + .send(); +---- + +TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`. + +This other example shows how to add a `ProducerInterceptor` that will intercept and mutate messages received by the producer before being published to the brokers: +[source, java] +---- +template.newMessage(msg) + .withProducerCustomizer((pb) -> pb.intercept(interceptor)) + .send(); +---- + +The customizer will only apply to the producer used for the send operation. +If you want to apply a customizer to all producers, you must provide them to the producer factory as described in <>. + +CAUTION: The rules described in "`<>`" must be followed when using Lambda customizers. + + +[[schema-info-template-imperative]] +:template-class: PulsarTemplate +include::../schema-info/schema-info-template.adoc[] + +[[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 which you can further configure by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.*`] application properties. + +NOTE: If topic information is not specified when using the producer factory APIs directly, the same xref:reference/pulsar/topic-resolution.adoc#topic-resolution-process-imperative[topic resolution process] used by the `PulsarTemplate` is used with the one exception that the "Message type default" step is **omitted**. + +[[global-producer-customize]] +=== Global producer customization +The framework provides the `ProducerBuilderCustomizer` contract which allows you to configure the underlying builder which is used to construct each producer. +To customize all producers, you can pass a list of customizers into the `PulsarProducerFactory` constructor. +When using multiple customizers, they are applied in the order in which they appear in the list. + +TIP: If you use Spring Boot auto-configuration, you can specify the customizers as beans and they will be passed automatically to the `PulsarProducerFactory`, ordered according to their `@Order` annotation. + +If you want to apply a customizer to just a single producer, you can use the Fluent API and <>. + +[[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-boot-pulsar-config-props}[`spring.pulsar.producer.cache.*`] application properties. + +[[producer-caching-lambdas]] +=== Caution on Lambda customizers +Any user-provided producer customizers are also included in the cache key. +Because the cache key relies on a valid implementation of `equals/hashCode`, one must take caution when using Lambda customizers. + +IMPORTANT: *RULE:* Two customizers implemented as Lambdas will match on `equals/hashCode` *if and only if* they use the same Lambda instance and do not require any variable defined outside its closure. + +To clarify the above rule we will look at a few examples. +In the following example, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. Additionally, it requires no variable outside its closure. Therefore, it *will* match as a cache key. + +[source, java] +---- +void sendUser() { + var user = randomUser(); + template.newMessage(user) + .withTopic("user-topic") + .withProducerCustomizer((b) -> b.producerName("user")) + .send(); +} +---- + +In this next case, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. However, it requires a variable outside its closure. Therefore, it *will not* match as a cache key. + +[source, java] +---- +void sendUser() { + var user = randomUser(); + var name = randomName(); + template.newMessage(user) + .withTopic("user-topic") + .withProducerCustomizer((b) -> b.producerName(name)) + .send(); +} +---- + +In this final example, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. While it does use a variable name, it does not originate outside its closure and therefore *will* match as a cache key. +This illustrates that variables can be used *within* the Lambda closure and can even make calls to static methods. + +[source, java] +---- +void sendUser() { + var user = randomUser(); + template.newMessage(user) + .withTopic("user-topic") + .withProducerCustomizer((b) -> { + var name = SomeHelper.someStaticMethod(); + b.producerName(name); + }) + .send(); +} +---- + +IMPORTANT: *RULE:* If your Lambda customizer is not defined *once and only once* (the same instance is used on subsequent calls) *OR* it requires variable(s) defined outside its closure then you must provide a customizer implementation with a valid `equals/hashCode` implementation. + +WARNING: If these rules are not followed then the producer cache will always miss and your application performance will be negatively affected. + +== Intercept Messages on the Producer +Adding a `ProducerInterceptor` lets you intercept and mutate messages received by the producer before they are published to the brokers. +To do so, you can pass a list of interceptors into the `PulsarTemplate` constructor. +When using multiple interceptors, the order they are applied in is the order in which they appear in the list. + +If you use Spring Boot auto-configuration, you can specify the interceptors as Beans. +They are passed automatically to the `PulsarTemplate`. +Ordering of the interceptors is achieved by using the `@Order` annotation as follows: + +[source, java] +---- +@Bean +@Order(100) +ProducerInterceptor firstInterceptor() { + ... +} + +@Bean +@Order(200) +ProducerInterceptor secondInterceptor() { + ... +} +---- + +NOTE: If you are not using the starter, you will need to configure and register the aforementioned components yourself. diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/preface.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/preface.adoc new file mode 100644 index 00000000..c8d6f9e8 --- /dev/null +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/preface.adoc @@ -0,0 +1,9 @@ +[[preface]] += Preface +include::../../attributes/attributes.adoc[] + +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-spring-boot-starter` module as a dependency. + +NOTE: The majority of this reference expects the reader to be using the starter and gives most directions for configuration with that in mind. +However, an effort is made to call out when instructions are specific to the Spring Boot starter usage. diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/publishing-consuming-partitioned-topics.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/publishing-consuming-partitioned-topics.adoc new file mode 100644 index 00000000..c122a7c9 --- /dev/null +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/publishing-consuming-partitioned-topics.adoc @@ -0,0 +1,111 @@ +[[publishing-consuming-partitioned-topics]] += Publishing and Consuming Partitioned Topics +include::../../attributes/attributes.adoc[] + +In the following example, we publish to a topic called `hello-pulsar-partitioned`. +It is a topic that is partitioned, and, for this sample, we assume that the topic is already created with three partitions. + +[source, java] +---- +@SpringBootApplication +public class PulsarBootPartitioned { + + public static void main(String[] args) { + SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.message-routing-mode=CustomPartition"); + } + + @Bean + public ApplicationRunner runner(PulsarTemplate 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; + } + } + +} +---- + +In the preceding example, we publish 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. +To do so, we provide a message router object with the `send` method. +Consider the three message routers implemented. +`FooRouter` always sends data to partition `0`, `BarRouter` sends to partition `1`, and `BuzzRouter` sends to partition `2`. +Also note that we now use 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.message-routing-mode`). + +On the consumer side, we use a `PulsarListener` with the exclusive subscription type. +This means that data from all the partitions ends 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: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = 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 = 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 = SubscriptionType.Failover) +public void listen3(String foo) { + System.out.println("Message Received 3: " + foo); +} +---- + +When you follow this approach, a single partition always gets consumed by a dedicated consumer. + +In a similar vein, if you want to use Pulsar's shared consumer type, you can use the `shared` subscription type. +However, when you use 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. + +Consider the following example: + +[source, java] +---- +@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = 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 = SubscriptionType.Shared) +public void listen2(String foo) { + System.out.println("Message Received 2: " + foo); +} +---- diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/pulsar-client.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/pulsar-client.adoc new file mode 100644 index 00000000..ec2b2029 --- /dev/null +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/pulsar-client.adoc @@ -0,0 +1,23 @@ +[[pulsar-client]] += Pulsar Client +include::../../attributes/attributes.adoc[] + +When you use the Pulsar Spring Boot Starter, you get the `PulsarClient` auto-configured. + +By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`. +This can be adjusted by setting the `spring.pulsar.client.service-url` property to a different value. + +TIP: The value must be a valid {apache-pulsar-docs}/client-libraries-java/#connection-urls[Pulsar Protocol] URL + +You can further configure the client by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application properties. + +NOTE: If you are not using the starter, you will need to configure and register the `PulsarClient` yourself. +There is a `DefaultPulsarClientFactory` that accepts a builder customizer that can be used to help with this. + +[[tls-encryption]] +== TLS Encryption (SSL) +include::tls-encryption.adoc[] + +[[client-authentication]] +== Authentication +include::authentication.adoc[] diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/quick-tour.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/quick-tour.adoc similarity index 98% rename from spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/quick-tour.adoc rename to spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/quick-tour.adoc index f63296fd..fe910a93 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/quick-tour.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/quick-tour.adoc @@ -1,6 +1,6 @@ [[quick-tour]] = Quick Tour -include::../attributes/attributes.adoc[] +include::../../attributes/attributes.adoc[] 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`. diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/tls-encryption.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/tls-encryption.adoc similarity index 97% rename from spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/tls-encryption.adoc rename to spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/tls-encryption.adoc index 18123604..cfa5a47a 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/tls-encryption.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/tls-encryption.adoc @@ -1,5 +1,4 @@ - -include::../attributes/attributes.adoc[] +include::../../attributes/attributes.adoc[] By default, Pulsar clients communicate with Pulsar services in plain text. The following section describes how to configure Pulsar clients to use TLS encryption (SSL). diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/topic-resolution.adoc similarity index 97% rename from spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc rename to spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/topic-resolution.adoc index 4f4f9049..bff4c052 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/topic-resolution.adoc @@ -1,3 +1,7 @@ +[[topic-resolution-process-imperative]] += Topic Resolution +include::../../attributes/attributes.adoc[] + A destination topic is needed when producing or consuming messages. The framework looks in the following ordered locations to determine a topic (stopping at the first find): diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc index acf701d6..3bc0d9f1 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc @@ -56,7 +56,7 @@ See the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application [[reactive-client-authentication]] === Authentication -To connect to a Pulsar cluster that requires authentication, follow xref:reference/pulsar.adoc#client-authentication[the same steps] as the imperative client. +To connect to a Pulsar cluster that requires authentication, follow xref:reference/pulsar/pulsar-client.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]] @@ -498,4 +498,4 @@ Spring Boot provides this reader factory which can be configured with any of the [[topic-resolution-process-reactive]] == Topic Resolution -include::topic-resolution.adoc[leveloffset=+1] +include::pulsar/topic-resolution.adoc[leveloffset=+1] diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc index 58fb7f6d..59d63486 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc @@ -8,13 +8,13 @@ This section covers the changes made from version 1.0 to version 1.1. === Auto Schema support If there is no chance to know the schema of a Pulsar topic in advance, you can use AUTO Schemas to produce/consume generic records to/from brokers. -See xref:./reference/pulsar.adoc#template-auto-produce[Producing with AUTO_SCHEMA] and xref:./reference/pulsar.adoc#listener-auto-consume[Consuming with AUTO_SCHEMA] for more details. +See xref:./reference/pulsar/message-production.adoc#template-auto-produce[Producing with AUTO_SCHEMA] and xref:./reference/pulsar/message-consumption.adoc#listener-auto-consume[Consuming with AUTO_SCHEMA] for more details. NOTE: While the above links focus on `PulsarTemplate` and `@PulsarListener`, this feature is also supported in `ReactivePulsarTemplate`, `@ReactivePulsarListener`, and `@PulsarReader`. Details for each can be found in their respective section of this reference guide. === Default topic/schema via message annotation -You can now mark a message class with `@PulsarMessage` to specify the xref:./reference/pulsar.adoc#default-topic-via-annotation[default topic] and/or xref:./reference/pulsar.adoc#listener-default-schema-annotation[default schema] to use when producing/consuming messages of that type. +You can now mark a message class with `@PulsarMessage` to specify the xref:./reference/pulsar/topic-resolution.adoc#default-topic-via-annotation[default topic] and/or xref:./reference/pulsar/message-consumption.adoc#listener-default-schema-annotation[default schema] to use when producing/consuming messages of that type. === Remove checked exceptions The APIs provided by the framework no longer throw the checked `PulsarClientException`, but rather the unchecked `PulsarException`.