Spring for Apache Pulsar

The template provides a handful of methods (prefixed with 'send') for simple send requests. For more complicated send requests, a fluent API lets you configure more options.

The template provides a fluent builder to handle more complicated send requests.

You can specify a TypedMessageBuilderCustomizer to configure the outgoing message. For example, the following code shows how to send a keyed message:

You can specify a ProducerBuilderCustomizer to configure the underlying Pulsar producer builder that ultimately constructs the producer used to send the outgoing message.

For example, the following code shows how to disable batching and enable chunking:

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:

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:

As an alternative to specifying the schema when invoking send operations on the PulsarTemplate for complex types, the schema resolver can be configured with mappings for the types. This removes the need to specify the schema as the framework consults the resolver using the outgoing message type.

Schema mappings can be configured with the spring.pulsar.defaults.type-mappings property. The following example uses application.yml to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can provide a schema resolver customizer to add the mapping(s).

The following example uses a schema resolver customizer to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

With this configuration in place, there is no need to set specify the schema on send operations.

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 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.pulsar.producer.cache.* application properties.

As an alternative to specifying the schema on the PulsarListener for complex types, the schema resolver can be configured with mappings for the types. This removes the need to set the schema on the listener as the framework consults the resolver using the incoming message type.

Schema mappings can be configured with the spring.pulsar.defaults.type-mappings property. The following example uses application.yml to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can provide a schema resolver customizer to add the mapping(s).

The following example uses a schema resolver customizer to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

With this configuration in place, there is no need to set the schema on the listener, for example:

This is a single consumer-based message listener container. The following listing shows its constructor:

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:

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:

DefaultPulsarMessageListenerContainer creates only a single consumer. If you want to have multiple consumers managed through multiple threads, you need to use ConcurrentPulsarMessageListenerContainer.

ConcurrentPulsarMessageListenerContainer has the following constructor:

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.

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.

The following listing shows another example of PulsarListener, but with Shared subscription and concurrency enabled.

In the preceding example, the PulsarListener creates five different consumers (this time, we assume that the topic has five partitions).

If you need message ordering and still want a shared subscription types, you need to use the Key_Shared subscription type.

Let us take a look at how the message listener container enables both single-record and batch-based message consumption.

Let us revisit our basic PulsarListener for the sake of this discussion:

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.

The following example shows the PulsarListener consuming records in batches:

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.

The following example shows how you can access the various Pulsar Headers in an application that uses the single record mode of consuming:

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.

In this section, we see how to access the various Pulsar Headers in an application that uses a batch consumer:

In the preceding example, we consume the data as a List<String>. 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.Message<?>, org.apache.pulsar.client.api.Messages<?>, or org.springframework.messaging.Messsge<?>.

Spring for Apache Pulsar provides the following modes for acknowledging messages:

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).

Let us revisit our basic single message based PulsarListener:

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:

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:

You can also set the listener property, spring.pulsar.listner.ack-mode, to set the ack mode application-wide. When doing this, you need not set this on the PulsarListener annotation. In that case, all the PulsarListener methods in the application acquire that property.

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:

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:

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:

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.

When you consume records in batches (see “Message ACK modes”) 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:

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.

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:

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:

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<MessageId>. You can also negatively acknowledge with the MessageId for the batch listener.

By default, Pulsar consumers does not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer. When you use Spring for Apache Pulsar, you can enable this property by setting the spring.pulsar.consumer.ack-timeout Boot property. If this property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered.

You can also specify this property directly as a Pulsar consumer property on the PulsarListener itself:

When you specify ackTimeout (as seen in the preceding PulsarListener method), 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:

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.

When acknowledging negatively, Pulsar consumer lets you specify how the application wants the message to be re-delivered. The default is to redeliver the message in one minute, but you can change it by setting spring.pulsar.consumer.negative-ack-redelivery-delay. You can also set it as a consumer property directly on PulsarListener, as follows:

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:

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:

First, we have a special bean for DeadLetterPolicy, and it is named as deadLetterPolicy (it can be any name as you wish). This bean specifies a number of things, such as the max delivery (10, in this case) and the name of the dead letter topic — my-dlq-topic, in this case. If you do not specify a DLQ topic name, it defaults to <topicname>-<subscriptionname>-DLQ in Pulsar. Next, we provide this bean name to 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.

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:

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:

The recovererForConsumer method takes a Pulsar consumer and returns a PulsarMessageRecoverer, which is another functional interface. Here is the API of PulsarMessageRecoverer:

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:

Note that we are provide a destination resolver to the PulsarDeadLetterPublishingRecoverer as the second constructor argument. If not provided, PulsarDeadLetterPublishingRecoverer uses <subscription-name>-<topic-name>-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.

First, let us look at a batch PulsarListener method:

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.

While it is possible to use PulsarReaderFactory directly, Spring for Apache Pulsar provides a convenient annotation called PulsarReader 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.

As you can see from the above example, PulsarReader is a quick way to read from a Pulsar topic. The id and subscriptionName attributes are optional, but always a best practice to provide them. PulsarReader requires topics and the startMessageId as mandatory attributes. 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.

You can customize any fields available through ReaderBuilder using a ReaderBuilderCustomizer in Spring for Apache Pulsar. You can provide a @Bean from ReaderBuilderCustomizer and then make it available to the PulsarReader as below.

The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can replace the default resolver by proving your own implementation, for example:

The template provides a fluent builder to handle more complicated send requests.

You can specify a MessageSpecBuilderCustomizer to configure the outgoing message. For example, the following code shows how to send a keyed message:

You can specify a ReactiveMessageSenderBuilderCustomizer to configure the underlying Pulsar sender builder that ultimately constructs the sender used to send the outgoing message.

For example, the following code shows how to disable batching and enable chunking:

This other example shows how to use custom routing when publishing records to partitioned topics. Specify your custom MessageRouter implementation on the sender builder such as:

As an alternative to specifying the schema when invoking send operations on the ReactivePulsarTemplate for complex types, the schema resolver can be configured with mappings for the types. This removes the need to specify the schema as the framework consults the resolver using the outgoing message type.

Schema mappings can be configured with the spring.pulsar.defaults.type-mappings property. The following example uses application.yml to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can provide a schema resolver customizer to add the mapping(s).

The following example uses a schema resolver customizer to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

With this configuration in place, there is no need to set specify the schema on send operations.

Each underlying Pulsar producer consumes resources. To improve performance and avoid continual creation of producers, the ReactiveMessageSenderCache in the underlying Apache Pulsar Reactive client caches the producers that it creates. They are cached in an LRU fashion and evicted when they have not been used within a configured time period.

You can configure the cache settings by specifying any of the spring.pulsar.reactive.sender.cache.* application properties.

All of the above are examples of consuming a single record one-by-one. However, one of the compelling reasons to use Reactive is for the streaming capability with backpressure support.

The following example uses ReactivePulsarListener to consume a stream of POJOs:

Here we receive the records as a Flux of messages. In addition, to enable stream consumption at the ReactivePulsarListener level, you need to set the stream property on the annotation to true.

Based on the actual type of the messages in the Flux, the framework tries to infer the schema to use. If it contains a complex type, you still need to provide the schemaType on ReactivePulsarListener.

The following listener uses the Spring messaging Message envelope with a complex type :

The listener ultimately relies on ReactivePulsarConsumerFactory to create and manage the underlying Pulsar consumer.

Spring Boot provides this consumer factory which can be configured with any of the spring.pulsar.reactive.consumer.* application-properties.

You can specify a ReactiveMessageConsumerBuilderCustomizer to configure the underlying Pulsar consumer builder that ultimately constructs the consumer used by the listener to receive the messages.

For example, the following code shows how to set the initial position of the subscription to the earliest messaage on the topic.

You can also use the customizer to provide direct Pulsar consumer properties to the consumer builder. This is convenient if you do not want to use the Boot configuration properties mentioned earlier or have multiple ReactivePulsarListener methods whose configuration varies.

The following customizer example uses direct Pulsar consumer properties:

As an alternative to specifying the schema on the ReactivePulsarListener for complex types, the schema resolver can be configured with mappings for the types. This removes the need to set the schema on the listener as the framework consults the resolver using the incoming message type.

Schema mappings can be configured with the spring.pulsar.defaults.type-mappings property. The following example uses application.yml to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can provide a schema resolver customizer to add the mapping(s).

The following example uses a schema resolver customizer to add mappings for the User and Address complex objects using AVRO and JSON schemas, respectively:

With this configuration in place, there is no need to set the schema on the listener, for example:

The contract for this message listener container is provided through ReactivePulsarMessageListenerContainer whose default implementation creates a reactive Pulsar consumer and wires up a reactive message pipeline that uses the created consumer.

The pipeline is a feature of the underlying Apache Pulsar Reactive client which does the heavy lifting of receiving the data in a reactive manner and then handing it over to the provided message handler. The reactive message listener container implementation is much simpler because the pipeline handles the majority of the work.

The "listener" aspect is provided by the ReactivePulsarMessageHandler of which there are two provided implementations:

The following example shows how you can access Pulsar Headers when using a one-by-one message listener:

In the preceding example, we access the values for the messageId message metadata as well as a custom message property named foo. The Spring @Header annotation is used for each header field.

You can also use Pulsar’s Message as the envelope to carry the payload. When doing so, the user can directly call the corresponding methods on the Pulsar message for retrieving the metadata. However, as a convenience, you can also retrieve it by using the Header annotation. Note that you can also use the Spring messaging Message envelope to carry the payload and then retrieve the Pulsar headers by using @Header.

When using a streaming message listener the header support is limited. Only when the Flux contains Spring org.springframework.messaging.Message elements will the headers be populated. Additionally, the Spring @Header annotation can not be used to retrieve the data. You must directly call the corresponding methods on the Spring message to retrieve the data.

The single message (aka OneByOne) message listener method returns a Mono<Void> to signal whether the message was successfully processed. Mono.empty() indicates success (acknowledgment) and Mono.error() indicates failure (negative acknowledgment).

The streaming listener method returns a Flux<MessageResult<Void>> where each MessageResult element represents a processed message and holds the message id, value and whether it was acknowledged. The MessageResult has a set of acknowledge and negativeAcknowledge static factory methods that can be used to create the appropriate MessageResult instance.

By default, Pulsar consumers do not redeliver messages unless the consumer crashes, but you can change this behavior by setting an ack timeout on the Pulsar consumer. When you use Spring for Apache Pulsar, you can enable this property by setting the spring.pulsar.reactive.consumer.ack-timeout Boot property. If this property has a value above zero and if the Pulsar consumer does not acknowledge a message within that timeout period, the message is redelivered.

You can also specify this property directly as a Pulsar consumer property via a consumer customizer such as:

When acknowledging negatively, Pulsar consumer lets you specify how the application wants the message to be re-delivered. The default is to redeliver the message in one minute, but you can change it by setting spring.pulsar.reactive.consumer.negative-ack-redelivery-delay.

You can also set it directly as a Pulsar consumer property via a consumer customizer such as:

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:

First, we have a special bean for DeadLetterPolicy, and it is named as deadLetterPolicy (it can be any name as you wish). This bean specifies a number of things, such as the max delivery (10, in this case) and the name of the dead letter topic — my-dlq-topic, in this case. If you do not specify a DLQ topic name, it defaults to <topicname>-<subscriptionname>-DLQ in Pulsar. Next, we provide this bean name to ReactivePulsarListener by setting the deadLetterPolicy property. Note that the ReactivePulsarListener has a subscription type of Shared, as the DLQ feature only works with shared subscriptions. This code is primarily for demonstration purposes, so we provide an ackTimeout value of 1 second. The idea is that the code throws the exception and, if Pulsar does not receive an ack within 1 second, it does a retry. If that cycle continues ten times (as that is our max redelivery count in the DeadLetterPolicy), the Pulsar consumer publishes the messages to the DLQ topic. We have another ReactivePulsarListener that listens on the DLQ topic to receive data as it is published to the DLQ topic.

The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can replace the default resolver by proving your own implementation, for example:

Metric name spring.pulsar.listener (defined by convention class org.springframework.pulsar.observation.DefaultPulsarListenerObservationConvention). Type timer.

Metric name spring.pulsar.listener.active (defined by convention class org.springframework.pulsar.observation.DefaultPulsarListenerObservationConvention). Type long task timer.

Fully qualified name of the enclosing class org.springframework.pulsar.observation.PulsarListenerObservation.

Metric name spring.pulsar.template (defined by convention class org.springframework.pulsar.observation.DefaultPulsarTemplateObservationConvention). Type timer.

Metric name spring.pulsar.template.active (defined by convention class org.springframework.pulsar.observation.DefaultPulsarTemplateObservationConvention). Type long task timer.

Fully qualified name of the enclosing class org.springframework.pulsar.observation.PulsarTemplateObservation.

Span name spring.pulsar.listener (defined by convention class org.springframework.pulsar.observation.DefaultPulsarListenerObservationConvention).

Fully qualified name of the enclosing class org.springframework.pulsar.observation.PulsarListenerObservation.

Span name spring.pulsar.template (defined by convention class org.springframework.pulsar.observation.DefaultPulsarTemplateObservationConvention).

Fully qualified name of the enclosing class org.springframework.pulsar.observation.PulsarTemplateObservation.

See Micrometer Tracing for more information.

The following example shows the steps to configure your Spring Boot application to use Zipkin with Brave.

At this point, your application should record traces when you send and receive Pulsar messages. You should be able to view them in the Zipkin UI (at localhost:9411, when running locally).

The steps are very similar to configuring any of the other supported Tracing environments.

Pulsar does not have a first-class “header” concept but instead provides a map for custom user properties as well as methods to access the message metadata typically stored in a message header (eg. id and event-time). As such, the terms “Pulsar message header” and “Pulsar message metadata” are used interchangeably. The list of available message metadata (headers) can be found in PulsarHeaders.java.

Spring Messaging provides first-class “header” support via its MessageHeaders abstraction.

The PulsarHeaderMapper strategy is provided to map headers to and from Pulsar user properties and Spring MessageHeaders.

Its interface definition is as follows:

The framework provides a couple of mapper implementations.

On the inbound side, by default, all Pulsar headers (message metadata plus user properties) are mapped to MessageHeaders. On the outbound side, by default, all MessageHeaders are mapped, except id, timestamp, and the headers that represent the Pulsar message metadata. You can specify which headers are mapped for inbound and outbound messages by configuring the inboundPatterns and outboundPatterns on a mapper bean you provide.

Patterns are rather simple and can contain a leading wildcard (*), a trailing wildcard, or both (for example, *.cat.*). You can negate patterns with a leading !. The first pattern that matches a header name (whether positive or negative) wins.

The Pulsar binder is configured with a default header mapper that can be overridden by providing your own PulsarHeaderMapper bean.

In the following example, a JSON header mapper is configured that: