Getting Started With Apache Pulsar and Spring Boot

1. Overview

Apache Pulsar is a distributed Publisher-Subscriber messaging system. While the features provided by Apache Pulsar are similar to those of Apache Kafka, Pulsar aims to overcome Kafka’s limitations of high latency, low throughput, difficulties in scaling and geo-replication, and more. Apache Pulsar is a good alternative when dealing with large volumes of data that require real-time processing.

In this tutorial, we’ll see how to integrate Apache Pulsar with our Spring Boot application. We’ll leverage the PulsarTemplate and PulsarListener configured by Pulsar’s Spring Boot Starter. We’ll also see how we can modify their default configurations according to our requirements.

2. Maven Dependency

We’ll first run a standalone Apache Pulsar server as described in Introduction to Apache Pulsar.

Next, let’s add the spring-pulsar-spring-boot-starter library to our project:

<dependency>
    <groupId>org.springframework.pulsar</groupId>
    <artifactId>spring-pulsar-spring-boot-starter</artifactId>
    <version>0.2.0</version>
</dependency>

3. PulsarClient

To interact with a Pulsar server, we need to configure a PulsarClient. By default, Spring auto-configures a PulsarClient that connects to the Pulsar server on localhost:6650:

spring:
  pulsar:
    client:
      service-url: pulsar://localhost:6650

We can change this configuration to establish a connection on a different address.

To connect to a secure server, we can use pulsar+ssl instead of pulsar. We can also configure properties like connection timeout, authentication, and memory limit, among others, by adding spring.pulsar.client.* properties to the application.yml.

4. Specifying Schema for Custom Object

We’ll use a simple User class for our application:

public class User {

    private String email;
    private String firstName;

    // standard constructors, getters and setters
}

Spring-Pulsar auto-detects primitive datatypes and generates the relevant schema. But, if we need to work with a custom JSON object, we’ll have to configure its schema information for the PulsarClient:

spring:
  pulsar:
    defaults:
      type-mappings:
        - message-type: com.baeldung.springpulsar.User
          schema-info:
            schema-type: JSON

Here, the message-type property accepts the fully qualified name of the message class and schema-type provides information about the schema type to be used. For complex objects, the schema-type property accepts either AVRO or JSON values.

Although using the properties file to specify the schema is the preferred method, we can also provide this schema through a bean:

@Bean
public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
    return (schemaResolver) -> {
        schemaResolver.addCustomSchemaMapping(User.class, Schema.JSON(User.class));
    }
}

This configuration should be added to both the producer as well as the listener applications.

5. Publisher

To publish messages on a Pulsar topic, we’ll use a PulsarTemplate. PulsarTemplate implements the PulsarOperations interface and provides methods to publish records in synchronous as well as asynchronous form. The send method blocks the calls to provide synchronous operation capabilities, whereas the sendAsync method offers non-blocking asynchronous operation.

In this tutorial, we’ll use the synchronous operation to publish records.

5.1. Publishing a Message

Spring Boot auto-configures a ready-to-use PulsarTemplate that publishes records to the specified topic.

Let’s create a producer that publishes String messages to the queue:

@Component
public class PulsarProducer {

    @Autowired
    private PulsarTemplate<String> stringTemplate;

    private static final String STRING_TOPIC = "string-topic";

    public void sendStringMessageToPulsarTopic(String str) throws PulsarClientException {
        stringTemplate.send(STRING_TOPIC, str);
    }
}

Now, let’s try to send a User object to a new queue:

@Autowired
private PulsarTemplate<User> template;

private static final String USER_TOPIC = "user-topic";

public void sendMessageToPulsarTopic(User user) throws PulsarClientException {
    template.send(USER_TOPIC, user);
}

In the above code snippet, we used the PulsarTemplate to send an object of the User class to Apache Pulsar’s topic called user-topic.

5.2. Producer-Side Customization

PulsarTemplate accepts TypedMessageBuilderCustomizer to configure outgoing messages and ProducerBuilderCustomizer to customize the producer’s properties.

We can use the TypedMessageBuilderCustomizer to configure message delay, send at a specific time, disable replication, and provide additional properties:

public void sendMessageToPulsarTopic(User user) throws PulsarClientException {
    template.newMessage(user)
      .withMessageCustomizer(mc -> {
        mc.deliverAfter(10L, TimeUnit.SECONDS);
      })
      .send();
}

ProducerBuilderCustomizer can be used to add an access mode, a custom message router, and an interceptor, and to enable or disable chunking and batching:

public void sendMessageToPulsarTopic(User user) throws PulsarClientException {
    template.newMessage(user)
      .withProducerCustomizer(pc -> {
        pc.accessMode(ProducerAccessMode.Shared);
      })
      .send();
}

6. Consumer

After publishing messages to our topic, we’ll now establish a listener for the same topic. To enable listening to a topic, we need to decorate the listener method with the @PulsarListener annotation.

Spring Boot configures all the necessary components for the listener method.

We also need to use @EnablePulsar to use the PulsarListener.

6.1. Receiving A Message

We’ll first create a listener method for the string-topic created in the earlier section:

@Service
public class PulsarConsumer {

    private static final String STRING_TOPIC = "string-topic";

    @PulsarListener(
      subscriptionName = "string-topic-subscription",
      topics = STRING_TOPIC,
      subscriptionType = SubscriptionType.Shared
    )
    public void stringTopicListener(String str) {
        LOGGER.info("Received String message: {}", str);
    }
}

Here, in the PulsarListener annotation, we’ve configured the topic this method would listen to in topicName and have given a subscription name in the subscriptionName attribute.

Now, let’s create a listener method for the user-topic used for the User class:

private static final String USER_TOPIC = "user-topic";

@PulsarListener(
    subscriptionName = "user-topic-subscription",
    topics = USER_TOPIC,
    schemaType = SchemaType.JSON
)
public void userTopicListener(User user) {
    LOGGER.info("Received user object with email: {}", user.getEmail());
}

Apart from the attributes provided in the earlier Listener method, we’ve also added a schemaType attribute that has the same value as that in its producer.

We’ll also add the @EnablePulsar annotation to our main class:

@EnablePulsar
@SpringBootApplication
public class SpringPulsarApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringPulsarApplication.class, args);
    }
}

6.2. Consumer-Side Customization

In addition to the subscription name and schema type, PulsarListener can be used to configure properties like auto-startup, batch, and acknowledgment mode:

@PulsarListener(
  subscriptionName = "user-topic-subscription",
  topics = USER_TOPIC,
  subscriptionType = SubscriptionType.Shared,
  schemaType = SchemaType.JSON,
  ackMode = AckMode.RECORD,
  properties = {"ackTimeout=60s"}
)
public void userTopicListener(User user) {
    LOGGER.info("Received user object with email: {}", user.getEmail());
}

Here, we’ve set the acknowledgment mode to Record and set the acknowledgment timeout to 60 seconds.

7. Using Dead-Letter Topic

If the acknowledgment of the message times out or the server receives nack, Pulsar tries to redeliver the message a certain number of times. After these retries are exhausted, these undelivered messages can be sent to queues known as Dead Letter Queues (DLQ).

This option is only available for the Shared subscription type. For configuring a DLQ for our user-topic queue, we’ll first create a DeadLetterPolicy bean that will define the number of times the redelivery should be attempted and the name of the queue to be used as DLQ:

private static final String USER_DEAD_LETTER_TOPIC = "user-dead-letter-topic";

@Bean
DeadLetterPolicy deadLetterPolicy() {
    return DeadLetterPolicy.builder()
      .maxRedeliverCount(10)
      .deadLetterTopic(USER_DEAD_LETTER_TOPIC)
      .build();
}

Now, we’ll add this policy to the PulsarListener we created earlier:

@PulsarListener(
  subscriptionName = "user-topic-subscription",
  topics = USER_TOPIC,
  subscriptionType = SubscriptionType.Shared,
  schemaType = SchemaType.JSON,
  deadLetterPolicy = "deadLetterPolicy",
  properties = {"ackTimeout=60s"}
)
public void userTopicListener(User user) {
    LOGGER.info("Received user object with email: {}", user.getEmail());
}

Here, we’ve configured the userTopicListener to use the deadLetterPolicy we created earlier, and we have configured an acknowledgment time of 60 seconds.

We can create a separate Listener to process the messages in the DQL:

@PulsarListener(
  subscriptionName = "dead-letter-topic-subscription",
  topics = USER_DEAD_LETTER_TOPIC,
  subscriptionType = SubscriptionType.Shared
)
public void userDlqTopicListener(User user) {
    LOGGER.info("Received user object in user-DLQ with email: {}", user.getEmail());
}

8. Conclusion

In this tutorial, we saw how to use Apache Pulsar with our Spring Boot applications and a few methods to change the default configurations.