Apache Pulsar consumer and producer with Spring Boot

Welcome to another Spring Boot tutorial. In this post, we will learn how to get started with Apache Pulsar on the Spring Boot ecosystem.

· Prerequisites
· Overview
∘ What is Apache Pulsar?
∘ Architecture
∘ Why use Apache Pulsar?
· Getting Started
∘ Case study
∘ Set up Apache Pulsar
∘ Apache Pulsar with Spring Boot
∘ Pulsar Producer
∘ Pulsar Consumer
· Test the Producer and Consumer
· Conclusion
· References

---

Prerequisites

This is the list of all the prerequisites:

• Spring Boot 3
• Maven 3.6.3 or later
• Java 21
• IntelliJ IDEA, Visual Studio Code, or another IDE
• Postman/insomnia or any other API testing tool.
• Docker installed
• Docker Compose is installed

Overview

What is Apache Pulsar?

Apache Pulsar is a cloud-native, multi-tenant, high-performance solution for server-to-server messaging and queuing built on the publisher-subscribe (pub-sub) pattern. Pulsar combines the best features of a traditional messaging system like RabbitMQ with those of a pub-sub system like Apache Kafka — scaling up or down dynamically without downtime. It’s used by thousands of companies for high-performance data pipelines, microservices, instant messaging, data integrations, and more.

Originally developed by Yahoo, Pulsar is under the stewardship of the Apache Software Foundation.

Architecture

A Pulsar instance is composed of one or more Pulsar clusters. This instance uses an instance-wide ZooKeeper cluster (configuration) for a store to retain information that pertains to multiple clusters, such as geo-replication
and tenant-level security policies.

Why use Apache Pulsar?

• It’s open-source ecosystem
• It’s structured to scale elastically.
• It provides extensive documentation.
• Pulsar offers multi-tenancy, which allows multiple groups of users to share the cluster.
• It offers management tooling built-in.
• Pulsar combines modularity and performance with multi-tenancy, geo-replication, consolidating Queueing and Streaming capabilities into one system.

Getting Started

Case study

Apache Pulsar supports client libraries in Java, Python, Go, C++, and C#. Spring Boot’s official Spring for Apache Pulsar integration. We’ll use the spring-boot-starter-pulsar dependency in a notification system example.

Suppose that an admin wants to send email newsletter notifications to users.

Set up Apache Pulsar

For local development and testing, we run Pulsar in standalone mode within a Docker container. We will create a docker-compose file containing all the instructions to run the Pulsar instance in standalone mode and the Pulsar Manager.

version: '3.8'

services:
   pulsar:
    image: apachepulsar/pulsar:latest
    container_name: pulsar
    command: >
      bin/pulsar standalone
    ports:
      - "6650:6650"   # Pulsar binary protocol port
      - "8080:8080"   # Pulsar HTTP service port
    environment:
      PULSAR_MEM: "-Xms512m -Xmx512m -XX:MaxDirectMemorySize=512m"
    volumes:
      - pulsar_data:/pulsar/data
      - pulsar_conf:/pulsar/conf
    networks:
      - pulsar_network    
      
   pulsar-manager:
    image: "apachepulsar/pulsar-manager:v0.4.0"
    container_name: pulsar-manager
    ports:
      - "9527:9527"
      - "7750:7750"
    depends_on:
      - pulsar
    links:
      - pulsar     
    environment:
      SPRING_CONFIGURATION_FILE: /pulsar-manager/pulsar-manager/application.properties
      LOG_LEVEL: DEBUG
      SERVER_PORT: 7750 
      
    networks:
      - pulsar_network   


volumes:
  pulsar_data:
  pulsar_conf:
  
networks:
  pulsar_network:
    name: pulsar_network
    driver: bridge    

Start the local standalone cluster:

docker-compose up -d

Once started, we can use any of the following URLs:

• pulsar://localhost:6650
• http://localhost:8080

Pulsar Manager UI

Pulsar Manager is a web-based GUI management and monitoring tool that helps administrators and users manage and monitor tenants, namespaces, topics, subscriptions, brokers, clusters, and so on, and supports dynamic configuration of multiple environments. — https://pulsar.apache.org/docs/administration-pulsar-manager

Access Pulsar manager UI at http://127.0.0.1:9527/

You can create a super-user using the following commands. Then you can use the super-user credentials to log in to the Pulsar Manager UI.

# Get CSRF token
CSRF_TOKEN=$(curl -s http://localhost:7750/pulsar-manager/csrf-token)

# Create super-user
curl \
  -H "X-XSRF-TOKEN: $CSRF_TOKEN" \
  -H "Cookie: XSRF-TOKEN=$CSRF_TOKEN;" \
  -H "Content-Type: application/json" \
  -X PUT http://localhost:7750/pulsar-manager/users/superuser \
  -d '{"name": "admin", "password": "apachepulsar", "description": "test", "email": "username@test.org"}'

Create an environment

An environment represents a Pulsar instance or a group of clusters you want to manage. A Pulsar Manager is capable of managing multiple environments.

Apache Pulsar with Spring Boot

Now, let’s start with the Spring Boot source code. We’ll start by creating two (2) Spring Boot projects from start.spring.io (spring-pulsar-producer and spring-pulsar-consumer).

Add the official Spring Boot Pulsar starter in your pom.xml:

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

Spring Boot will automatically configure the Pulsar client if you provide the connection URL.

Pulsar Producer

Producers are used to write messages to topics. We will use our spring-pulsar-producer app to publish notification messages to the consumer app.

The first step is to add producer properties in the application.yml file:

spring:
  pulsar:
    client:
      operation-timeout: 30m
      service-url: pulsar://127.0.0.1:6650
      # pulsar This configuration is not required if no password has been set.
      authentication:
        plugin-class-name: org.apache.pulsar.client.impl.auth.AuthenticationToken
        param:
          token: GeneratedTokenValue
    producer:
      topic-name: notification-topic
      name: notification_sender

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.

Define a Message Payload

Create a Notification class representing messages:

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public class Notification {

    @NotNull
    @JsonProperty("recipientEmail")
    private String recipientEmail;
    @NotNull
    @JsonProperty("recipientName")
    private String recipientName;
    @NotNull
    @JsonProperty("subject")
    private String subject;
    @NotNull
    @JsonProperty("channel")
    private String channel ;
    @NotNull
    @JsonProperty("content")
    private String content;
}

The following complex schema types are currently supported: JSON, AVRO, PROTOBUF, AUTO_PRODUCE_BYTES, and KEY_VALUE with INLINE encoding.

Create a Pulsar Producer

For our case study, we have a REST API that receives the newsletter the user wants to send to subscribers and stores it in the database. Then select all subscribers and send the messages to the broker.

Spring Boot auto-configuration provides a PulsarTemplate for publishing records. The template implements an interface called PulsarOperations and offers methods to publish records through its contract.

@Slf4j
@RequiredArgsConstructor
@Service
@Transactional
public class NewsletterServiceImpl implements NewsletterService {

    private final NewsletterRepository repository;
    private final NewsletterSubscriberService newsletterSubscriberService;

    private final PulsarTemplate<Object> pulsarTemplate;

    /**
     * {@inheritDoc}
     */
    @Override
    public Newsletter sendNewsletter(NewsletterDTO newsletterDTO) {

        Newsletter newsletter = Newsletter.builder()
                .channel("EMAIL").content(newsletterDTO.getContent())
                .title(newsletterDTO.getTitle()).createdAt(LocalDateTime.now())
                .publishedAt(LocalDateTime.now()).type(newsletterDTO.getType())
                .build();

        Newsletter newsletterSaved = repository.save(newsletter);
        if (ObjectUtils.isNotEmpty(newsletterSaved)) {
            sendToSubscriber(newsletterSaved);
        }

        return newsletterSaved;
    }

    private void sendToSubscriber(Newsletter newsletterSaved) {

            // get all suscribers
            List<NewsletterSubscriber> newsletterSubscribers = newsletterSubscriberService.getAll();

            if(!newsletterSubscribers.isEmpty()){
                // send email notification
                newsletterSubscribers.forEach(newsletterSubscriber -> {
                    Notification notification = Notification.builder()
                            .channel(newsletterSaved.getChannel()).content(newsletterSaved.getContent())
                            .recipientEmail(newsletterSubscriber.getEmail()).recipientName(newsletterSubscriber.getName())
                            .subject(newsletterSaved.getTitle())
                            .build();

                    sendMsgAsync(notification);
                });
        }
    }

    /**
     * Pulsar publish messages asynchronously using the Java client
     */
    private void sendMsgAsync(Notification notification) {
        pulsarTemplate.sendAsync(notification).thenAccept(msgId -> log.info("Notification message with ID {} successfully sent", msgId));
    }
}

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.

Our producer project structure looks like this:

Pulsar Consumer

Consumers subscribe to topics and handle messages that producers publish to those topics. After a message has been successfully processed, an acknowledgment should be sent back to the broker to indicate that we are done processing the message within the subscription.

Consumer properties in the application.yml file:

spring:
  pulsar:
    client:
      operation-timeout: 30m
      service-url: pulsar://127.0.0.1:6650
    consumer:
      name: notification_sender
      topics:
        - notification-topic

Create a Pulsar Consumer

When it comes to Pulsar consumers, it is recommended that end-user applications use the PulsarListener annotation. To use PulsarListener, you need to use the @EnablePulsar annotation. In 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).

@Slf4j
@Component
public class NotificationMessageListener{

    @PulsarListener(
            ackMode = AckMode.MANUAL,
            subscriptionType = SubscriptionType.Shared)
    public void received(Message<Notification> msg, Acknowledgement ack) {
        try {
            log.info("Topic Name: {}", msg.getTopicName());
            log.info("Message Id: {}", msg.getMessageId());
            log.info("Producer Name: {}", msg.getProducerName());
            log.info("Publish Time: {}", msg.getPublishTime());

            //log.info("Message received: {}", new String(msg.getData()));
            Notification notification = msg.getValue();
            log.info("Message received => Username: {}, Email: {}, Subject: {}, content: {}", notification.getRecipientName(),
                    notification.getRecipientEmail(), notification.getSubject(), StringUtils.abbreviate(notification.getContent(), 100));

            log.info("####################################################################################");
            ack.acknowledge(msg.getMessageId());
        } catch (Exception e) {
            ack.nack();
        }
    }

}

The received() method is called whenever a new message is received. Messages are guaranteed to be delivered in order and from the same thread for a single consumer.

Message<Notification> msg: The actual message received from Pulsar, with the payload deserialized as a Notification object.

You can consume the Pulsar message directly:

org.apache.pulsar.client.api.Message<Notification> msg

or consume by using the Spring messaging envelope:

org.springframework.messaging.Message<Notification> msg

For more details and a full list of configuration options, you can refer to the official Spring Boot Pulsar documentation:
Spring Boot Pulsar — Message Consumption

Test the Producer and Consumer

Run the producer and consumer applications.

• Producer:

• Consumer

Pulsar Manager UI

Conclusion

Well done !!. In this post, we explored how to integrate Apache Pulsar with the Spring Boot ecosystem, including setting up producers and consumers with manual acknowledgment.

When deciding between messaging systems like Pulsar, RabbitMQ, or Kafka, your choice should depend on your specific requirements. Each system has its own strengths and trade-offs in terms of scalability, performance, and feature set.

The key is to choose the configuration that best fits your use case to ensure reliable and efficient message processing.

Support me through GitHub Sponsors.

Thank you for reading!! See you in the next post.

Last updated: November 2025

References

• https://pulsar.apache.org/
• https://github.com/apache/pulsar-manager
• https://docs.spring.io/spring-pulsar/docs/1.2.11/reference/index.html
• https://streamnative.io/en/blog/tech/2020-11-09-benchmark-pulsar-kafka-performance/

👉 Link to Medium blog