Spring for Apache Pulsar

The minimum supported versions for the underlying libraries required by the framework are as follows:

Spring Boot applications only need the spring-pulsar-spring-boot-starter dependency. The following shows how to define the dependency for Maven and Gradle, respectively:

Let us go through the higher-level details of this application quickly. Later on in this documentation, we will see these components in much more detail.

In the sample above, we are heavily relying on Spring Boot auto-configuration. Spring Boot auto-configures several components for our application. It automatically provides a PulsarClient for the application which is used by both the producer and the consumer.

PulsarTemplate also is auto-configured by Spring Boot which we inject in the application and start sending records to a Pulsar topic. The application sends messages to a topic named hello-pulsar. Note that the application does not specify any schema information. That is because Spring for Apache Pulsar library automatically infers the schema type from the type of the data that you are sending.

We use PulsarListener annotation to consume from the hello-pulsar topic where we publish the data. PulsarListener is a convenient annotation that wraps the message listener container infrastructure in Spring for Apache Pulsar. Behind the scenes, it creates a message listener container which creates and manages the Pulsar consumer. As with a regular Pulsar consumer, the default subscription type when using PulsarListener is the Exclusive mode. As records are published in to the hello-pulsar topic, the Pulsarlistener consumes them and prints them on the console. Here also, the framework infers the schema type used from the data type that the PulsarListner method uses as the payload - String in this case.

When using the Pulsar Spring Boot Starter, you get the PulsarClient auto-configured. This is done through a factory bean called PulsarClientFactoryBean, which takes a configuration object PulsarClientConfiguration. By default, the application tries to connect to a local Pulsar instance at pulsar://localhost:6650. However, there are many application properties available to configure the client.

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 are blocking calls using the synchronous sending capabilities on the Pulsar producer. They return the MessageId of the message that was published once the message is persisted on the broker. The sendAsync method calls are asynchronous calls that are non-blocking. They return a CompletableFuture using which you can asynchronously receive the message id once the messages are published.

The PulsarTemplate relies on a PulsarProducerFactory for actually creating the underlying producer. Spring Boot auto-configuration also provides this producer factory. Additionally, you can configure the factory by specifying any of the available producer-centric application properties listed above.

Each underlying Pulsar producer consumes resources. In order 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.producer.cache prefixed application properties listed above.

When it comes to Pulsar consumer, we recommend the end user applications to make use of the PulsarListener annotation. In order to use PulsarListener, you need to use the @EnablePulsar annotation. When using the Spring Boot support, it automatically enables this annotation and configures all the components necessary for PulsarListener such as the message listener infrastructure which is responsible for creating the Pulsar consumer. PulsarMessageListenerContainer uses a PulsarConsumerFactory in order to create and manage the Pulsar consumer. This consumer factory is also auto-configured through Spring Boot.

Let us revisit the PulsarListener code snippet we saw in the quick-tour section.

This can even be further simplified as below.

In this most basic form, you must provide the following two properties with their corresponding values.

In the PulsarListener method above, we receive the data as String, but we don’t specify any schema types. Internally, the framework relies on Pulsar’s schema mechanism to convert the data to the required type. The framework detects that you are expecting the String type and then infers the schema type based on that information. Then it provides that schema to the consumer. For all the primitive types in Java, the framework does this inference. For any complex types, such as JSON, AVRO etc. the framework cannot do this inference and the user needs to provide the schema type on the annotation using the schemaType property.

Here is another PulsarListener method, that takes an Integer.

The following PulsarListener method shows how we can consume complex types from a topic

Note the addition of a schemaType property on PulsarListener. That is because the library is not capable of inferring the schema type from the provided type Foo, we must tell the framework what schema to use.

Here is an example of using PulsarListener to consume records in batches.

Note that in this example, we are receiving the records as a collection (List) of objects. In addition, in order to enable batch consumption at the PulsarListener level, you need to set the batch property on the annotation to true.

Based on the actual type that the List holds, the framework tries to infer the schema to use. If the List contains a complex type, then the schemaType still needs to be provided on PulsarListener.

The following also should work in which we use the Message envelope provided by the Pulsar Java client.

When using 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 above or have multiple PulsarListener methods.

Here is an example of using Pulsar consumer properties directly on PulsarListener.

Note that the properties used are direct Pulsar consumer properties.

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 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 and this section will go through those details.

As briefly mentioned above, the message listener container is at the heart of message consumption when using Spring for Apache Pulsar. PulsarListener uses the message listener container infrastructure behind the scenes to create and manage the Pulsar consumer. Spring for Apache Pulsar provides the contract for this message listener container through PulsarMessageListenerContainer. The default implementation for this message listener container is provided through DefaultPulsarMessageListenerContainer. As its name indicates, PulsarMessageListenerContainer contains the message listener. The container creates the Pulsar consumer and then runs a separate thread to receive and handle the data. The data is handled by the provided message listener implementation.

The message listener container consumes the data in batch using the consumer’s batchReceive method. Once data is received, it is handed over to the selected message listener implementation.

The following message listener types are available when using Spring for Apache Pulsar.

We will see the details about these various message listeners in the sections below.

Before doing so however, lets take a closer look at the container itself

In this section, we are going to see how the message listener container enables both single record and batch based message consumption.

Let us re-visit our basic PulsarListener for the sake of this discussion.

With this PulsarListener method, what we are essentially doing is that asking Spring for Apache Pulsar to invoke the listener method with a single record each time. We mentioned that the message listener container consumes the data in batches using the batchReceive method on the consumer. The framework detects that the PulsarListener in this case receives a single record which means that on each invocation of the method it needs a singe record. Although the records are consumed by the message listener container in batches, it iterates through the received batch and then invoke the listener method through an adapter for PulsarRecordMessageListener. As you can see in the previous section, PulsarRecordMessageListener simply extends from the MessageListener provided by the Pulsar Java client and it supports the basic received method.

Here is the PulsarListener example of consuming records in batches.

When using this type of PulsarListener, the framework detects that you are in batch mode. Since it is already received the data in batches using the Consumer’s batchReceive method, it simply hands off the entire batch to the listener method through an adapter for PulsarBatchMessageListener.

When using Spring for Apache Pulsar, the message acknowledgment is handled by the framework unless opted out by the application. In this section, we go through the details of how the framework takes care of message acknowledgment.

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 will see how acknowledgment works when using both single and batch versions of PulsarListener and how they translate to the backing message listener container (and of course ultimately to the Pulsar consumer).

Let us revisit our basic single message based PulsarListener.

It is natural to wonder, how acknowledgment works when using PulsarListener, espcially if you are familiar with Pulsar consumer directly. The answer comes down to the message listener container as that is the central place in Spring for Apache Pulsar which coordinates all the consumer related activities.

Assuming you are not overriding the default behavior, this is what happens behind the scenes when using the above PulsarListener.

This is the normal flow. If any record from the original batch received, throws an exception, Spring for Apache Pulsar will track them separately. When all the records from the batch are processed, then Spring for Apache Pulsar will acknowledge all the succesful messages and negatively acknowledge (nack) all the failed messages. In other words, when consuming single records using PulsarRecordMessageListener and the default ack mode of BATCH is used, the framework waits for all the record received from the batchReceive call to process successfully and then call the acknowledge method on the Pulsar Consumer. If any particular record throws an exception when invoking the handler method, Spring for Apache Pulsar tracks those records and separately call negativeAcknowledge on those records after the entire batch is processed.

If the application wants the acknowledgment or negative acknowledgment to occur per record, then the RECORD ack mode can be enabled. In that case, after handling each record, the message is acknowledged if no error and negatively acknowledged if there was an error. Here is an example of enabling RECORD ack mode on 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 do not need to set this on the PulsarListener annotation. In that case, all the PulsarListener methods in the application acquires that property.

There are situations in which you might not want the framework to do any acknowledgments, but rather do that directly from the application itself. Spring for Apache Pulsar provides a couple of ways to enable manual message acknowledgments. Let us look at a few examples.

Few things merit explanation here - First we are enabling manual ack mode by setting ackMode on PulsarListener. When enabling manual ack mode, Spring for Apache Pulsar allows the application to inject an Acknowledgment object as you can see in the above PulsarListener method. 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 to your PulsarListener while using MANUAL ack mode and then call one of the corresponding methods above.

In the above PulsarListener example, we are calling a parameter-less acknowledge method. This is because the framework knows which Message it is operating under currently. When calling acknowledge(), you do not need to receive the payload with the Message enveloper`, but rather simply using 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 using acknowledge(messageId), you must receive the payload using the Message<?> envelope.

Similar to what is possible for acknowledging, the Acknowledgment API also provides options for negatively acknowledging - see the nack methods above.

You can also call acknowledge directly on the Pulsar consumer as below.

As you can see, when calling acknowledge directly on the underlying consumer, then you need to do error handling by yourself. Using the Acknowledgment does not require that as the framework can do that for you. Therefore, it is recommended to use the Acknowledgment object approach when using manual acknowledgment.

When using manual acknowledgment, it is important to understand that the framework completely stay from any acknowledgment at all. Hence, it is extremely important for the end-users to think through the right acknowledgment strategies when designing applications.

When records are consumed in batches (See the section above), then if the default ack mode of BATCH is used, then when the entire batch is processed successfully, it will be acknowledged. If any records throw an exception, then the entire batch is negatively acknowledged. 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

Let us look at the following batch listener:

When all the messages in the incoming collection (messages in this example) are processed, the framework will acknowledge all of them.

When consuming in batch mode, RECORD is not an allowed ack mode. This might cause an issue as application may not want the entire batch to be re-delivered again. For 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, then the framework will not do any acknowledgment - positive or negative. It is entirely up to the application to take care of such concerns. When MANUAL ack mode is set, Spring for Apache Pulsar selects a compatible message listener container - PulsarBatchAcknowledgingMessageListener for batch consumption which gives you access to an Acknowledgment object. Once again, the following are the methods availble in the Acknowledgment API.

You can inject this Acknowledgment object to your PulsarListener while using MANUAL ack mode. Here is a basic example for a batch based listener.

When using a batch listener, the message listener container cannot know which record it is currently operating upon. Therefore, in order to manually acknowledge, you need to use one of the overloaded acknowledge method that takes a MessageId or a List<MessageId>. You can also negatively acknowledge with the MessageId for the batch listener.

In the sample below, we are publishing to a topic called hello-pulsar-partitioned. It is a topic that is partitioned and for this sample we assume that the topic is already created with three partitions.

A few things require explanation in the application above. We are publishing to a partitioned topic and we would like to publish some data segment to a specific partition. If you leave it to Pulsar’s default, it follows a round-robin mode of partition assignments, and we would like to override that. In order to do that, we are providing a message router object with the send method. Look at the three message routers implemented. FooRouter always sends data to partition 0, BarRouter to partition 1 and BuzzRouter to partition 2. Also note that, we are now using the sendAsync method of PulsarTemplate that returns a CompletableFuture. When running the application, we also need to set the messageRoutingMode on the producer to CustomPartition (spring.pulsar.producer.message-routing-mode).

On the consumer side, we are using a PulsarListener with the exclusive subscription type. This means that data from all the partitions will end up in the same consumer and there is no ordering guarantee.

What can we do if we want each partition to be consumed by a single distinct consumer? We can switch to the failover subscription mode and add three separate consumers.

Here is an example.

When following this approach, you can see that a single partition always gets consumed by a dedicated consumer.

In the similar vein, if you want to use Pulsar’s shared consumer type, you can use the subscription type shared. Keep in mind though, that when using the shared mode, you lose any ordering guarantees as a single consumer may receive messages from all the partitions before another consumer gets a chance.

Here is an example.

In your PulsarListener method, you can receive the record directly as a Pulsar Message instead of the actual payload type. Here is an example.

or in batch receiver:

Sometimes, it is necessary to gain direct access to the Pulsar Consumer object. Here is how you may do so.

When accessing the Consumer object this way, make sure NOT to invoke any operations that would change the Consumer’s cursor position by invoking any receive methods. All such operations must be done by the container.

As indicated above, for normal Java types (the primitive ones), Spring Pulsar framework can infer the proper Schema to use on the PulsarListener. However, for more complex types such as JSON or AVRO, you need to specify the schema type on the annotation. Here is how you provide that.

On the producer side also, for the Java primitive types, the framework can infer the Schema, but for any other types, you need to set them on the PulsarTemplate as shown below.

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, and we are going to take a look at them first and see how we can leverage them through Spring for Apache Pulsar.

On the Pulsar administration side, Spring Boot auto-configuration provides a PulsarAdministration to manage Pulsar clusters. The administration implements an interface called PulsarAdminOperations and provides a 'createOrModify' method to handle topic administration through its contract.

When using the Pulsar Spring Boot Starter, you get the PulsarAdministration auto-configured. By default, the application tries to connect to a local Pulsar instance at localhost:8080. However, there are many application properties available to configure the client.

On initialization, the PulsarAdministration checks if there are any PulsarTopic beans in the application context. For all such beans, the PulsarAdministration will either create the corresponding topic, or if necessary modify the number of partitions.

Below is an example how to add PulsarTopic beans to let the PulsarAdministration auto-create topics for you.

The reference documentation has the following appendices: