Pulsar Java client

You can use a Pulsar Java client to create the Java producer, consumer, reader and TableView of messages and to perform administrative tasks. The current Java client version is 2.11.2.

All the methods in producer, consumer, reader and TableView of a Java client are thread-safe.

Javadoc for the Pulsar client is divided into two domains by package as follows.

This document focuses only on the client API for producing and consuming messages on Pulsar topics. For how to use the Java admin client, see Pulsar admin interface.

Installation

The latest version of the Pulsar Java client library is available via Maven Central. To use the latest version, add the pulsar-client library to your build configuration.

tip

• pulsar-client and pulsar-client-admin shade dependencies via maven-shade-plugin to avoid conflicts of the underlying dependency packages (such as Netty). If you do not want to manage dependency conflicts manually, you can use them.
• pulsar-client-original and pulsar-client-admin-original does not shade dependencies. If you want to manage dependencies manually, you can use them.

Maven

If you use Maven, add the following information to the pom.xml file.

<!-- in your <properties> block -->
<pulsar.version>2.11.2</pulsar.version>
<!-- in your <dependencies> block -->
<dependency>
  <groupId>org.apache.pulsar</groupId>
  <artifactId>pulsar-client</artifactId>
  <version>${pulsar.version}</version>
</dependency>

Gradle

If you use Gradle, add the following information to the build.gradle file.

def pulsarVersion = '2.11.2'
dependencies {
    compile group: 'org.apache.pulsar', name: 'pulsar-client', version: pulsarVersion
}

Connection URLs

To connect to Pulsar using client libraries, you need to specify a Pulsar protocol URL.

You can assign Pulsar protocol URLs to specific clusters and use the pulsar scheme. The following is an example of localhost with the default port 6650:

pulsar://localhost:6650

If you have multiple brokers, separate IP:port by commas:

pulsar://localhost:6550,localhost:6651,localhost:6652

If you use mTLS authentication, add +ssl in the scheme:

pulsar+ssl://pulsar.us-west.example.com:6651

Client

You can instantiate a PulsarClient object using just a URL for the target Pulsar cluster like this:

PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650")
        .build();

If you have multiple brokers, you can initiate a PulsarClient like this:

PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650,localhost:6651,localhost:6652")
        .build();

note

If you run a cluster in standalone mode, the broker is available at the pulsar://localhost:6650 URL by default.

If you create a client, you can use the loadConf configuration. The following parameters are available in loadConf.

Check out the Javadoc for the PulsarClient class for a full list of configurable parameters.

In addition to client-level configuration, you can also apply producer and consumer specific configuration as described in sections below.

Client memory allocator configuration

You can set the client memory allocator configurations through Java properties.

Example

Dpulsar.allocator.pooled=true
Dpulsar.allocator.exit_on_oom=false
Dpulsar.allocator.leak_detection=Disabled
Dpulsar.allocator.out_of_memory_policy=ThrowException

Producer

In Pulsar, producers write messages to topics. Once you’ve instantiated a PulsarClient object (as in the section above), you can create a Producer for a specific Pulsar topic.

Producer<byte[]> producer = client.newProducer()
        .topic("my-topic")
        .create();
// You can then send messages to the broker and topic you specified:
producer.send("My message".getBytes());

By default, producers produce messages that consist of byte arrays. You can produce different types by specifying a message schema.

Producer<String> stringProducer = client.newProducer(Schema.STRING)
        .topic("my-topic")
        .create();
stringProducer.send("My message");

Make sure that you close your producers, consumers, and clients when you do not need them.

producer.close();
consumer.close();
client.close();

Close operations can also be asynchronous:

producer.closeAsync()
   .thenRun(() -> System.out.println("Producer closed"))
   .exceptionally((ex) -> {
       System.err.println("Failed to close producer: " + ex);
       return null;
   });

Configure producer

If you instantiate a Producer object by specifying only a topic name as the example above, the default configuration of producer is used.

If you create a producer, you can use the loadConf configuration. The following parameters are available in loadConf.

You can configure parameters if you do not want to use the default configuration.

For a full list, see the Javadoc for the ProducerBuilder class. The following is an example.

Producer<byte[]> producer = client.newProducer()
    .topic("my-topic")
    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
    .sendTimeout(10, TimeUnit.SECONDS)
    .blockIfQueueFull(true)
    .create();

Publish to partitioned topics

By default, Pulsar topics are served by a single broker, which limits the maximum throughput of a topic. Partitioned topics can span multiple brokers and thus allow for higher throughput.

You can publish messages to partitioned topics using Pulsar client libraries. When publishing messages to partitioned topics, you must specify a routing mode. If you do not specify any routing mode when you create a new producer, the round-robin routing mode is used.

Routing mode

You can specify the routing mode in the ProducerConfiguration object used to configure your producer. The routing mode determines which partition (internal topic) each message should be published to.

The following MessageRoutingMode options are available.

The following is an example:

String pulsarBrokerRootUrl = "pulsar://localhost:6650";
String topic = "persistent://my-tenant/my-namespace/my-topic";
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsarBrokerRootUrl).build();
Producer<byte[]> producer = pulsarClient.newProducer()
        .topic(topic)
        .messageRoutingMode(MessageRoutingMode.SinglePartition)
        .create();
producer.send("Partitioned topic message".getBytes());

Custom message router

To use a custom message router, you need to provide an implementation of the MessageRouter interface, which has just one choosePartition method:

public interface MessageRouter extends Serializable {
    int choosePartition(Message msg);
}

The following router routes every message to partition 10:

public class AlwaysTenRouter implements MessageRouter {
    public int choosePartition(Message msg) {
        return 10;
    }
}

With that implementation, you can send messages to partitioned topics as below.

String pulsarBrokerRootUrl = "pulsar://localhost:6650";
String topic = "persistent://my-tenant/my-cluster-my-namespace/my-topic";
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsarBrokerRootUrl).build();
Producer<byte[]> producer = pulsarClient.newProducer()
        .topic(topic)
        .messageRouter(new AlwaysTenRouter())
        .create();
producer.send("Partitioned topic message".getBytes());

How to choose partitions when using a key

If a message has a key, it supersedes the round robin routing policy. The following example illustrates how to choose the partition when using a key.

// If the message has a key, it supersedes the round robin routing policy
        if (msg.hasKey()) {
            return signSafeMod(hash.makeHash(msg.getKey()), topicMetadata.numPartitions());
        }
        if (isBatchingEnabled) { // if batching is enabled, choose partition on `partitionSwitchMs` boundary.
            long currentMs = clock.millis();
            return signSafeMod(currentMs / partitionSwitchMs + startPtnIdx, topicMetadata.numPartitions());
        } else {
            return signSafeMod(PARTITION_INDEX_UPDATER.getAndIncrement(this), topicMetadata.numPartitions());
        }

Async send

You can publish messages asynchronously using the Java client. With async send, the producer puts the message in a blocking queue and returns it immediately. Then the client library sends the message to the broker in the background. If the queue is full (max size configurable), the producer is blocked or fails immediately when calling the API, depending on arguments passed to the producer.

The following is an example.

producer.sendAsync("my-async-message".getBytes()).thenAccept(msgId -> {
    System.out.println("Message with ID " + msgId + " successfully sent");
});

As you can see from the example above, async send operations return a MessageId wrapped in a CompletableFuture.

Configure messages

In addition to a value, you can set additional items on a given message:

producer.newMessage()
    .key("my-message-key")
    .value("my-async-message".getBytes())
    .property("my-key", "my-value")
    .property("my-other-key", "my-other-value")
    .send();

You can terminate the builder chain with sendAsync() and get a future return.

Enable chunking

Message chunking enables Pulsar to process large payload messages by splitting the message into chunks at the producer side and aggregating chunked messages on the consumer side.

The message chunking feature is OFF by default. The following is an example of how to enable message chunking when creating a producer.

Producer<byte[]> producer = client.newProducer()
        .topic(topic)
        .enableChunking(true)
        .enableBatching(false)
        .create();

By default, producer chunks the large message based on max message size (maxMessageSize) configured at broker (eg: 5MB). However, client can also configure max chunked size using producer configuration chunkMaxMessageSize.

note

To enable chunking, you need to disable batching (enableBatching\=false) concurrently.

Intercept messages

ProducerInterceptor intercepts and possibly mutates messages received by the producer before they are published to the brokers.

The interface has three main events:

• eligible checks if the interceptor can be applied to the message.
• beforeSend is triggered before the producer sends the message to the broker. You can modify messages within this event.
• onSendAcknowledgement is triggered when the message is acknowledged by the broker or the sending failed.

To intercept messages, you can add a ProducerInterceptor or multiple ones when creating a Producer as follows.

Producer<byte[]> producer = client.newProducer()
        .topic(topic)
        .intercept(new ProducerInterceptor {
            @Override
            boolean eligible(Message message) {
                return true;  // process all messages
            }
            @Override
            Message beforeSend(Producer producer, Message message) {
                // user-defined processing logic
            }
            @Override
            void onSendAcknowledgement(Producer producer, Message message, MessageId msgId, Throwable exception) {
                // user-defined processing logic
            }
        })
        .create();

note

Multiple interceptors apply in the order they are passed to the intercept method.

Consumer

In Pulsar, consumers subscribe to topics and handle messages that producers publish to those topics. You can instantiate a new consumer by first instantiating a PulsarClient object and passing it a URL for a Pulsar broker (as above).

Once you’ve instantiated a PulsarClient object, you can create a Consumer by specifying a topic and a subscription.

Consumer consumer = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscribe();

The subscribe method will auto-subscribe the consumer to the specified topic and subscription. One way to make the consumer listen to the topic is to set up a while loop. In this example loop, the consumer listens for messages, prints the contents of any received message, and then acknowledges that the message has been processed. If the processing logic fails, you can use negative acknowledgment to redeliver the message later.

while (true) {
  // Wait for a message
  Message msg = consumer.receive();
  try {
      // Do something with the message
      System.out.println("Message received: " + new String(msg.getData()));
      // Acknowledge the message
      consumer.acknowledge(msg);
  } catch (Exception e) {
      // Message failed to process, redeliver later
      consumer.negativeAcknowledge(msg);
  }
}

If you don’t want to block your main thread but constantly listen for new messages, consider using a MessageListener. The MessageListener uses a thread pool inside the client. You can set the number of threads for message listeners in the ClientBuilder.

MessageListener myMessageListener = (consumer, msg) -> {
  try {
      System.out.println("Message received: " + new String(msg.getData()));
      consumer.acknowledge(msg);
  } catch (Exception e) {
      consumer.negativeAcknowledge(msg);
  }
}
Consumer consumer = client.newConsumer()
     .topic("my-topic")
     .subscriptionName("my-subscription")
     .messageListener(myMessageListener)
     .subscribe();

Configure consumer

If you instantiate a Consumer object by specifying only a topic and subscription name as in the example above, the consumer uses the default configuration.

When you create a consumer, you can use the loadConf configuration. The following parameters are available in loadConf.

You can configure parameters if you do not want to use the default configuration. For a full list, see the Javadoc for the ConsumerBuilder class.

The following is an example.

Consumer consumer = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .ackTimeout(10, TimeUnit.SECONDS)
        .subscriptionType(SubscriptionType.Exclusive)
        .subscribe();

Async receive

The receive method receives messages synchronously (the consumer process is blocked until a message is available). You can also use async receive, which returns a CompletableFuture object immediately once a new message is available.

The following is an example.

CompletableFuture<Message> asyncMessage = consumer.receiveAsync();

Async receive operations return a Message wrapped inside of a CompletableFuture.

Batch receive

Use batchReceive to receive multiple messages for each call.

The following is an example.

Messages messages = consumer.batchReceive();
for (Object message : messages) {
  // do something
}
consumer.acknowledge(messages)

note

Batch receive policy limits the number and bytes of messages in a single batch. You can specify a timeout to wait for enough messages. The batch receive is completed if any of the following conditions are met: enough number of messages, bytes of messages, wait timeout.

Consumer consumer = client.newConsumer()
.topic("my-topic")
.subscriptionName("my-subscription")
.batchReceivePolicy(BatchReceivePolicy.builder()
.maxNumMessages(100)
.maxNumBytes(1024 * 1024)
.timeout(200, TimeUnit.MILLISECONDS)
.build())
.subscribe();

The default batch receive policy is:

BatchReceivePolicy.builder()
.maxNumMessage(-1)
.maxNumBytes(10 * 1024 * 1024)
.timeout(100, TimeUnit.MILLISECONDS)
.build();

Configure chunking

You can limit the maximum number of chunked messages a consumer maintains concurrently by configuring the maxPendingChunkedMessage and autoAckOldestChunkedMessageOnQueueFull parameters. When the threshold is reached, the consumer drops pending messages by silently acknowledging them or asking the broker to redeliver them later. The expireTimeOfIncompleteChunkedMessage parameter decides the time interval to expire incomplete chunks if the consumer fails to receive all chunks of a message within the specified time period.

The following is an example of how to configure message chunking.

Consumer<byte[]> consumer = client.newConsumer()
        .topic(topic)
        .subscriptionName("test")
        .autoAckOldestChunkedMessageOnQueueFull(true)
        .maxPendingChunkedMessage(100)
        .expireTimeOfIncompleteChunkedMessage(10, TimeUnit.MINUTES)
        .subscribe();

Negative acknowledgment redelivery backoff

The RedeliveryBackoff introduces a redelivery backoff mechanism. You can achieve redelivery with different delays by setting redeliveryCount of messages.

Consumer consumer =  client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .negativeAckRedeliveryBackoff(MultiplierRedeliveryBackoff.builder()
                .minDelayMs(1000)
                .maxDelayMs(60 * 1000)
                .build())
        .subscribe();

Acknowledgment timeout redelivery backoff

The RedeliveryBackoff introduces a redelivery backoff mechanism. You can redeliver messages with different delays by setting the number of times the messages are retried.

Consumer consumer =  client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .ackTimeout(10, TimeUnit.SECOND)
        .ackTimeoutRedeliveryBackoff(MultiplierRedeliveryBackoff.builder()
                .minDelayMs(1000)
                .maxDelayMs(60000)
                .multiplier(2)
                .build())
        .subscribe();

The message redelivery behavior should be as follows.

note

• The negativeAckRedeliveryBackoff does not work with consumer.negativeAcknowledge(MessageId messageId) because you are not able to get the redelivery count from the message ID.
• If a consumer crashes, it triggers the redelivery of unacked messages. In this case, RedeliveryBackoff does not take effect and the messages might get redelivered earlier than the delay time from the backoff.

Multi-topic subscriptions

In addition to subscribing a consumer to a single Pulsar topic, you can also subscribe to multiple topics simultaneously using multi-topic subscriptions. To use multi-topic subscriptions you can supply either a regular expression (regex) or a List of topics. If you select topics via regex, all topics must be within the same Pulsar namespace.

The followings are some examples.

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
ConsumerBuilder consumerBuilder = pulsarClient.newConsumer()
        .subscriptionName(subscription);
// Subscribe to all topics in a namespace
Pattern allTopicsInNamespace = Pattern.compile("public/default/.*");
Consumer allTopicsConsumer = consumerBuilder
        .topicsPattern(allTopicsInNamespace)
        .subscribe();
// Subscribe to a subsets of topics in a namespace, based on regex
Pattern someTopicsInNamespace = Pattern.compile("public/default/foo.*");
Consumer allTopicsConsumer = consumerBuilder
        .topicsPattern(someTopicsInNamespace)
        .subscribe();

In the above example, the consumer subscribes to the persistent topics that can match the topic name pattern. If you want the consumer subscribes to all persistent and non-persistent topics that can match the topic name pattern, set subscriptionTopicsMode to RegexSubscriptionMode.AllTopics.

Pattern pattern = Pattern.compile("public/default/.*");
pulsarClient.newConsumer()
        .subscriptionName("my-sub")
        .topicsPattern(pattern)
        .subscriptionTopicsMode(RegexSubscriptionMode.AllTopics)
        .subscribe();

note

By default, the subscriptionTopicsMode of the consumer is PersistentOnly. Available options of subscriptionTopicsMode are PersistentOnly, NonPersistentOnly, and AllTopics.

You can also subscribe to an explicit list of topics (across namespaces if you wish):

List<String> topics = Arrays.asList(
        "topic-1",
        "topic-2",
        "topic-3"
);
Consumer multiTopicConsumer = consumerBuilder
        .topics(topics)
        .subscribe();
// Alternatively:
Consumer multiTopicConsumer = consumerBuilder
        .topic(
            "topic-1",
            "topic-2",
            "topic-3"
        )
        .subscribe();

You can also subscribe to multiple topics asynchronously using the subscribeAsync method rather than the synchronous subscribe method. The following is an example.

Pattern allTopicsInNamespace = Pattern.compile("persistent://public/default.*");
consumerBuilder
        .topics(topics)
        .subscribeAsync()
        .thenAccept(this::receiveMessageFromConsumer);
private void receiveMessageFromConsumer(Object consumer) {
    ((Consumer)consumer).receiveAsync().thenAccept(message -> {
                // Do something with the received message
                receiveMessageFromConsumer(consumer);
            });
}

Subscription types

Pulsar has various subscription types to match different scenarios. A topic can have multiple subscriptions with different subscription types. However, a subscription can only have one subscription type at a time.

A subscription is identical to the subscription name; a subscription name can specify only one subscription type at a time. To change the subscription type, you should first stop all consumers of this subscription.

Different subscription types have different message distribution types. This section describes the differences between subscription types and how to use them.

To better describe their differences, assume you have a topic named “my-topic”, and the producer has published 10 messages.

Producer<String> producer = client.newProducer(Schema.STRING)
        .topic("my-topic")
        .enableBatching(false)
        .create();
// 3 messages with "key-1", 3 messages with "key-2", 2 messages with "key-3" and 2 messages with "key-4"
producer.newMessage().key("key-1").value("message-1-1").send();
producer.newMessage().key("key-1").value("message-1-2").send();
producer.newMessage().key("key-1").value("message-1-3").send();
producer.newMessage().key("key-2").value("message-2-1").send();
producer.newMessage().key("key-2").value("message-2-2").send();
producer.newMessage().key("key-2").value("message-2-3").send();
producer.newMessage().key("key-3").value("message-3-1").send();
producer.newMessage().key("key-3").value("message-3-2").send();
producer.newMessage().key("key-4").value("message-4-1").send();
producer.newMessage().key("key-4").value("message-4-2").send();

Exclusive

Create a new consumer and subscribe with the Exclusive subscription type.

Consumer consumer = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Exclusive)
        .subscribe()

Only the first consumer is allowed to the subscription, other consumers receive an error. The first consumer receives all 10 messages, and the consuming order is the same as the producing order.

note

If topic is a partitioned topic, the first consumer subscribes to all partitioned topics, other consumers are not assigned with partitions and receive an error.

Failover

Create new consumers and subscribe with theFailover subscription type.

Consumer consumer1 = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Failover)
        .subscribe()
Consumer consumer2 = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Failover)
        .subscribe()
//conumser1 is the active consumer, consumer2 is the standby consumer.
//consumer1 receives 5 messages and then crashes, consumer2 takes over as an  active consumer.

Multiple consumers can attach to the same subscription, yet only the first consumer is active, and others are standby. When the active consumer is disconnected, messages will be dispatched to one of standby consumers, and the standby consumer then becomes the active consumer.

If the first active consumer is disconnected after receiving 5 messages, the standby consumer becomes active consumer. Consumer1 will receive:

("key-1", "message-1-1")
("key-1", "message-1-2")
("key-1", "message-1-3")
("key-2", "message-2-1")
("key-2", "message-2-2")

consumer2 will receive:

("key-2", "message-2-3")
("key-3", "message-3-1")
("key-3", "message-3-2")
("key-4", "message-4-1")
("key-4", "message-4-2")

note

If a topic is a partitioned topic, each partition has only one active consumer, messages of one partition are distributed to only one consumer, and messages of multiple partitions are distributed to multiple consumers.

Shared

Create new consumers and subscribe with Shared subscription type.

Consumer consumer1 = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Shared)
        .subscribe()
Consumer consumer2 = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Shared)
        .subscribe()
//Both consumer1 and consumer2 are active consumers.

In Shared subscription type, multiple consumers can attach to the same subscription and messages are delivered in a round-robin distribution across consumers.

If a broker dispatches only one message at a time, consumer1 receives the following information.

("key-1", "message-1-1")
("key-1", "message-1-3")
("key-2", "message-2-2")
("key-3", "message-3-1")
("key-4", "message-4-1")

consumer2 receives the following information.

("key-1", "message-1-2")
("key-2", "message-2-1")
("key-2", "message-2-3")
("key-3", "message-3-2")
("key-4", "message-4-2")

The Shared subscription is different from the Exclusive and Failover subscription types. Shared subscription has better flexibility, but cannot provide an ordering guarantee.

Key_shared

This is a new subscription type since 2.4.0 release. Create new consumers and subscribe with Key_Shared subscription type.

Consumer consumer1 = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Key_Shared)
        .subscribe()
Consumer consumer2 = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Key_Shared)
        .subscribe()
//Both consumer1 and consumer2 are active consumers.

Just like in the Shared subscription, all consumers in the Key_Shared subscription type can attach to the same subscription. But the Key_Shared subscription type is different from the Shared subscription. In the Key_Shared subscription type, messages with the same key are delivered to only one consumer in order. The possible distribution of messages between different consumers (by default we do not know in advance which keys will be assigned to a consumer, but a key will only be assigned to a consumer at the same time).

consumer1 receives the following information.

("key-1", "message-1-1")
("key-1", "message-1-2")
("key-1", "message-1-3")
("key-3", "message-3-1")
("key-3", "message-3-2")

consumer2 receives the following information.

("key-2", "message-2-1")
("key-2", "message-2-2")
("key-2", "message-2-3")
("key-4", "message-4-1")
("key-4", "message-4-2")

If batching is enabled at the producer side, messages with different keys are added to a batch by default. The broker will dispatch the batch to the consumer, so the default batch mechanism may break the Key_Shared subscription guaranteed message distribution semantics. The producer needs to use the KeyBasedBatcher.

Producer producer = client.newProducer()
        .topic("my-topic")
        .batcherBuilder(BatcherBuilder.KEY_BASED)
        .create();

Or the producer can disable batching.

Producer producer = client.newProducer()
        .topic("my-topic")
        .enableBatching(false)
        .create();

note

If the message key is not specified, messages without keys are dispatched to one consumer in order by default.

Intercept messages

ConsumerInterceptors intercept and possibly mutate messages received by the consumer.

The interface has six main events:

• beforeConsume is triggered before the message is returned by receive() or receiveAsync(). You can modify messages within this event.
• onAcknowledge is triggered before the consumer sends the acknowledgement to the broker.
• onAcknowledgeCumulative is triggered before the consumer sends the cumulative acknowledgement to the broker.
• onNegativeAcksSend is triggered when a redelivery from a negative acknowledgement occurs.
• onAckTimeoutSend is triggered when a redelivery from an acknowledgement timeout occurs.
• onPartitionsChange is triggered when the partitions of the (partitioned) topic change.

To intercept messages, you can add one or multiple ConsumerInterceptors when creating a Consumer as follows.

Consumer<String> consumer = client.newConsumer()
        .topic("my-topic")
        .subscriptionName("my-subscription")
        .intercept(new ConsumerInterceptor<String> {
              @Override
              public Message<String> beforeConsume(Consumer<String> consumer, Message<String> message) {
                  // user-defined processing logic
              }
              @Override
              public void onAcknowledge(Consumer<String> consumer, MessageId messageId, Throwable cause) {
                  // user-defined processing logic
              }
              @Override
              public void onAcknowledgeCumulative(Consumer<String> consumer, MessageId messageId, Throwable cause) {
                  // user-defined processing logic
              }
              @Override
              public void onNegativeAcksSend(Consumer<String> consumer, Set<MessageId> messageIds) {
                  // user-defined processing logic
              }
              @Override
              public void onAckTimeoutSend(Consumer<String> consumer, Set<MessageId> messageIds) {
                  // user-defined processing logic
              }
              @Override
              public void onPartitionsChange(String topicName, int partitions) {
                  // user-defined processing logic
              }
        })
        .subscribe();

note

If you are using multiple interceptors, they apply in the order they are passed to the intercept method.

Reader

With the reader interface, Pulsar clients can “manually position” themselves within a topic and read all messages from a specified message onward. The Pulsar API for Java enables you to create Reader objects by specifying a topic and a MessageId.

The following is an example.

byte[] msgIdBytes = // Some message ID byte array
MessageId id = MessageId.fromByteArray(msgIdBytes);
Reader reader = pulsarClient.newReader()
        .topic(topic)
        .startMessageId(id)
        .create();
while (true) {
    Message message = reader.readNext();
    // Process message
}

In the example above, a Reader object is instantiated for a specific topic and message (by ID); the reader iterates over each message in the topic after the message is identified by msgIdBytes (how that value is obtained depends on the application).

The code sample above shows pointing the Reader object to a specific message (by ID), but you can also use MessageId.earliest to point to the earliest available message on the topic of MessageId.latest to point to the most recent available message.

Configure reader

When you create a reader, you can use the loadConf configuration. The following parameters are available in loadConf.

Sticky key range reader

In a sticky key range reader, broker only dispatches messages which hash of the message key contains by the specified key hash range. Multiple key hash ranges can be specified on a reader.

The following is an example to create a sticky key range reader.

pulsarClient.newReader()
        .topic(topic)
        .startMessageId(MessageId.earliest)
        .keyHashRange(Range.of(0, 10000), Range.of(20001, 30000))
        .create();

The total hash range size is 65536, so the max end of the range should be less than or equal to 65535.

Configure chunking

Configuring chunking for readers is similar to that for consumers. See configure chunking for consumers for more information.

The following is an example of how to configure message chunking for a reader.

Reader<byte[]> reader = pulsarClient.newReader()
        .topic(topicName)
        .startMessageId(MessageId.earliest)
        .maxPendingChunkedMessage(12)
        .autoAckOldestChunkedMessageOnQueueFull(true)
        .expireTimeOfIncompleteChunkedMessage(12, TimeUnit.MILLISECONDS)
        .create();

Create reader with interceptor

Pulsar reader interceptor intercepts and possibly mutates messages with user-defined processing before Pulsar reader reads them. With reader interceptors, you can apply unified messaging processes before messages can be read, such as modifying messages, adding properties, collecting statistics and etc, without creating similar mechanisms respectively.

Pulsar reader interceptor works on top of Pulsar consumer interceptor. The plugin interface ReaderInterceptor can be treated as a subset of ConsumerInterceptor and it has two main events.

• beforeRead is triggered before readers read messages. You can modify messages within this event.
• onPartitionsChange is triggered when changes on partitions have been detected.

To perceive triggered events and perform customized processing, you can add ReaderInterceptor when creating a Reader as follows.

PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build();
Reader<byte[]> reader = pulsarClient.newReader()
        .topic("t1")
        .autoUpdatePartitionsInterval(5, TimeUnit.SECONDS)
        .intercept(new ReaderInterceptor<byte[]>() {
            @Override
            public void close() {
            }
            @Override
            public Message<byte[]> beforeRead(Reader<byte[]> reader, Message<byte[]> message) {
                // user-defined processing logic
                return message;
            }
            @Override
            public void onPartitionsChange(String topicName, int partitions) {
                // user-defined processing logic
            }
        })
        .startMessageId(MessageId.earliest)
        .create();

TableView

The TableView interface serves an encapsulated access pattern, providing a continuously updated key-value map view of the compacted topic data. Messages without keys will be ignored.

With TableView, Pulsar clients can fetch all the message updates from a topic and construct a map with the latest values of each key. These values can then be used to build a local cache of data. In addition, you can register consumers with the TableView by specifying a listener to perform a scan of the map and then receive notifications when new messages are received. Consequently, event handling can be triggered to serve use cases, such as event-driven applications and message monitoring.

note

Each TableView uses one Reader instance per partition, and reads the topic starting from the compacted view by default. It is highly recommended to enable automatic compaction by configuring the topic compaction policies for the given topic or namespace. More frequent compaction results in shorter startup times because less data is replayed to reconstruct the TableView of the topic. Starting from Pulsar 2.11.0, TableView also supports reading non-persistent topics, but it does not guarantee data consistency.

The following figure illustrates the dynamic construction of a TableView updated with newer values of each key.

Configure TableView

The following is an example of how to configure a TableView.

TableView<String> tv = client.newTableViewBuilder(Schema.STRING)
  .topic("my-tableview")
  .create()

You can use the available parameters in the loadConf configuration or related API to customize your TableView.

Register listeners

You can register listeners for both existing messages on a topic and new messages coming into the topic by using forEachAndListen, and specify to perform operations for all existing messages by using forEach.

The following is an example of how to register listeners with TableView.

// Register listeners for all existing and incoming messages
tv.forEachAndListen((key, value) -> /*operations on all existing and incoming messages*/)
// Register action for all existing messages
tv.forEach((key, value) -> /*operations on all existing messages*/)

Schema

In Pulsar, all message data consists of byte arrays “under the hood.” Message schemas enable you to use other types of data when constructing and handling messages (from simple types like strings to more complex, application-specific types). If you construct, say, a producer without specifying a schema, then the producer can only produce messages of type byte[]. The following is an example.

Producer<byte[]> producer = client.newProducer()
        .topic(topic)
        .create();

The producer above is equivalent to a Producer<byte[]> (in fact, you should always explicitly specify the type). If you’d like to use a producer for a different type of data, you need to specify a schema that informs Pulsar which data type will be transmitted over the topic. For more examples, see Schema - Get started.

Authentication

Pulsar Java clients currently support the following authentication mechansims:

• mTLS
• JWT
• Athenz
• Kerberos
• OAuth2
• HTTP basic

Cluster-level failover

For more concepts and reference information about cluster-level failover, including concepts, benefits, use cases, constraints, usage and working principles, see Cluster-level failover.

tip

• You should configure cluster-level failover only when the cluster contains sufficient resources to handle all possible consequences. Workload intensity on the backup cluster may increase significantly.

• Connect clusters to an uninterruptible power supply (UPS) unit to reduce the risk of unexpected power loss.

Requirements

• Pulsar client 2.10 or later versions.

• For backup clusters:

  • The number of BookKeeper nodes should be equal to or greater than the ensemble quorum.

  • The number of ZooKeeper nodes should be equal to or greater than 3.

• Turn on geo-replication between the primary cluster and any dependent cluster (primary to backup or backup to backup) to prevent data loss.

• Enable replicated subscription.

• Automatic cluster-level failover

• Controlled cluster-level failover

This is an example of how to construct a Java Pulsar client to use automatic cluster-level failover. The switchover is triggered automatically.

private PulsarClient getAutoFailoverClient() throws PulsarClientException {
    String primaryUrl = "pulsar+ssl://localhost:6651";
    String secondaryUrl = "pulsar+ssl://localhost:6661";
    String primaryTlsTrustCertsFilePath = "primary/path";
    Authentication primaryAuthentication = AuthenticationFactory.create(
        "org.apache.pulsar.client.impl.auth.AuthenticationTls",
        "tlsCertFile:/path/to/primary-my-role.cert.pem,"
                + "tlsKeyFile:/path/to/primary-role.key-pk8.pem");
    String secondaryTlsTrustCertsFilePath = "secondary/path";
    Authentication secondaryAuthentication = AuthenticationFactory.create(
        "org.apache.pulsar.client.impl.auth.AuthenticationTls",
        "tlsCertFile:/path/to/secondary-my-role.cert.pem,"
                + "tlsKeyFile:/path/to/secondary-role.key-pk8.pem");
    // You can put more failover cluster config in to map
    Map<String, String> secondaryTlsTrustCertsFilePaths = new HashMap<>();
    secondaryTlsTrustCertsFilePaths.put(secondaryUrl, secondaryTlsTrustCertsFilePath);
    Map<String, Authentication> secondaryAuthentications = new HashMap<>();
    secondaryAuthentications.put(secondaryUrl, secondaryAuthentication);
    ServiceUrlProvider failover = AutoClusterFailover.builder()
        .primary(primaryUrl)
        .secondary(List.of(secondaryUrl))
        .failoverDelay(30, TimeUnit.SECONDS)
        .switchBackDelay(60, TimeUnit.SECONDS)
        .checkInterval(1000, TimeUnit.MILLISECONDS)
        .secondaryTlsTrustCertsFilePath(secondaryTlsTrustCertsFilePaths)
        .secondaryAuthentication(secondaryAuthentications)
        .build();
    PulsarClient pulsarClient = PulsarClient.builder()
        .authentication(primaryAuthentication) 
        .tlsTrustCertsFilePath(primaryTlsTrustCertsFilePath)
        .build();
    failover.initialize(pulsarClient);
    return pulsarClient;
}

Configure the following parameters:

This is an example of how to construct a Java Pulsar client to use controlled cluster-level failover. The switchover is triggered by administrators manually.

Note: you can have one or several backup clusters but can only specify one.

public PulsarClient getControlledFailoverClient() throws IOException {
    Map<String, String> header = new HashMap();
    header.put("service_user_id", "my-user");
    header.put("service_password", "tiger");
    header.put("clusterA", "tokenA");
    header.put("clusterB", "tokenB");
    ServiceUrlProvider provider = ControlledClusterFailover.builder()
        .defaultServiceUrl("pulsar://localhost:6650")
        .checkInterval(1, TimeUnit.MINUTES)
        .urlProvider("http://localhost:8080/test")
        .urlProviderHeader(header)
        .build();
    PulsarClient pulsarClient = PulsarClient.builder().build();
    provider.initialize(pulsarClient);
    return pulsarClient;
}

Here is an example of how urlProviderHeader works.

Assume that you want to connect Pulsar client 1 to cluster A.

1. Pulsar client 1 sends the token t1 to the URL provider service.

2. The URL provider service returns the credential c1 and the cluster A URL to the Pulsar client.

   The URL provider service manages all tokens and credentials. It returns different credentials based on different tokens and different target cluster URLs to different Pulsar clients.

   Note: the credential must be in a JSON file and contain parameters as shown.

   {
   "serviceUrl": "pulsar+ssl://target:6651",
   "tlsTrustCertsFilePath": "/security/ca.cert.pem",
   "authPluginClassName":"org.apache.pulsar.client.impl.auth.AuthenticationTls",
   "authParamsString": " \"tlsCertFile\": \"/security/client.cert.pem\"
       \"tlsKeyFile\": \"/security/client-pk8.pem\" "
   }

3. Pulsar client 1 connects to cluster A using credential c1.