The Pulsar Java client

    Javadoc for the Pulsar client is divided up into two domains, by package:

    This document will focus only on the client API for producing and consuming messages on Pulsar topics. For a guide to using the Java admin client, see The Pulsar admin interface.

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

    If you're using Maven, add this to your pom.xml:

    Gradle

    If you're using Gradle, add this to your build.gradle file:

    1. def pulsarVersion = '2.4.0'
    3. dependencies {
    4.     compile group: 'org.apache.pulsar', name: 'pulsar-client', version: pulsarVersion
    5. }

    Connection URLs

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

    Pulsar protocol URLs are assigned to specific clusters, use the pulsar scheme and have a default port of 6650. Here's an example for localhost:

    1. pulsar://localhost:6650

    If you have more than one broker, the URL may look like this:

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

    A URL for a production Pulsar cluster may look something like this:

    1. pulsar://pulsar.us-west.example.com:6650

    If you're using authentication, the URL will look like something like this:

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

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

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

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

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

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

    In addition to client-level configuration, you can also apply and consumer specific configuration, as you'll see in the sections below.

    Producers

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

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

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

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

    Configuring producers

    If you instantiate a Producer object specifying only a topic name, as in the example above, the producer will use the default configuration. To use a non-default configuration, there's a variety of configurable parameters that you can set. For a full listing, see the Javadoc for the ProducerBuilder class. Here's an example:

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

    Message routing

    When using partitioned topics, you can specify the routing mode whenever you publish messages using a producer. For more on specifying a routing mode using the Java client, see the Partitioned Topics cookbook.

    You can also publish messages using the Java client. With async send, the producer will put the message in a blocking queue and return immediately. The client library will then send the message to the broker in the background. If the queue is full (max size configurable), the producer could be blocked or fail immediately when calling the API, depending on arguments passed to the producer.

    Here's an example async send operation:

    1. producer.sendAsync("my-async-message".getBytes()).thenAccept(msgId -> {
    2.     System.out.printf("Message with ID %s successfully sent", msgId);
    3. });

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

    Configuring messages

    In addition to a value, it's possible to set additional items on a given message:

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

    Once you've instantiated a PulsarClient object, you can create a by specifying a topic and a .

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

    The subscribe method will automatically subscribe the consumer to the specified topic and subscription. One way to make the consumer listen on the topic is to set up a while loop. In this example loop, the consumer listens for messages, prints the contents of any message that's received, and then acknowledges that the message has been processed. If the processing logic fails, we use to have the message redelivered at a later point in time.

    1. while (true) {
    2.   // Wait for a message
    3.   Message msg = consumer.receive();
    5.   try {
    6.       // Do something with the message
    7.       System.out.printf("Message received: %s", new String(msg.getData()));
    9.       // Acknowledge the message so that it can be deleted by the message broker
    10.       consumer.acknowledge(msg);
    11.   } catch (Exception e) {
    12.       // Message failed to process, redeliver later
    13.       consumer.negativeAcknowledge(msg);
    14.   }
    15. }

    Configuring consumers

    If you instantiate a Consumer object specifying only a topic and subscription name, as in the example above, the consumer will use the default configuration. To use a non-default configuration, there's a variety of configurable parameters that you can set. For a full listing, see the Javadoc for the class. Here's an example:

    Here's an example configuration:

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

    Async receive

    The receive method will receive messages synchronously (the consumer process will be blocked until a message is available). You can also use , which will return immediately with a CompletableFuture object that completes once a new message is available.

    Here's an example:

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

    Async receive operations return a wrapped inside of a CompletableFuture.

    In addition to subscribing a consumer to a single Pulsar topic, you can also subscribe to multiple topics simultaneously using . 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.

    Here are some examples:

    1. import org.apache.pulsar.client.api.Consumer;
    2. import org.apache.pulsar.client.api.PulsarClient;
    4. import java.util.Arrays;
    5. import java.util.List;
    6. import java.util.regex.Pattern;
    8. ConsumerBuilder consumerBuilder = pulsarClient.newConsumer()
    9.         .subscriptionName(subscription);
    11. Pattern allTopicsInNamespace = Pattern.compile("persistent://public/default/.*");
    12. Consumer allTopicsConsumer = consumerBuilder
    13.         .topicsPattern(allTopicsInNamespace)
    14.         .subscribe();
    16. // Subscribe to a subsets of topics in a namespace, based on regex
    17. Pattern someTopicsInNamespace = Pattern.compile("persistent://public/default/foo.*");
    18. Consumer allTopicsConsumer = consumerBuilder
    19.         .topicsPattern(someTopicsInNamespace)
    20.         .subscribe();

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

    1. List<String> topics = Arrays.asList(
    2.         "topic-1",
    3.         "topic-2",
    4.         "topic-3"
    7. Consumer multiTopicConsumer = consumerBuilder
    8.         .topics(topics)
    9.         .subscribe();
    11. // Alternatively:
    12. Consumer multiTopicConsumer = consumerBuilder
    13.         .topics(
    14.             "topic-1",
    15.             "topic-2",
    16.             "topic-3"
    17.         )
    18.         .subscribe();

    You can also subscribe to multiple topics asynchronously using the subscribeAsync method rather than the synchronous subscribe method. Here's an example:

    1. Pattern allTopicsInNamespace = Pattern.compile("persistent://public/default.*");
    2. consumerBuilder
    3.         .topics(topics)
    4.         .subscribeAsync()
    5.         .thenAccept(this::receiveMessageFromConsumer);
    7. private void receiveMessageFromConsumer(Consumer consumer) {
    8.     consumer.receiveAsync().thenAccept(message -> {
    9.                 // Do something with the received message
    10.                 receiveMessageFromConsumer(consumer);
    11.             });
    12. }

    Subscription modes

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

    A subscription is identified with the subscription name, and a subscription name can specify only one subscription mode at a time. You can change the subscription mode, yet you have to let all existing consumers of this subscription offline first.

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

    In order to better describe their differences, assuming you have a topic named "my-topic", and the producer has published 10 messages.

    1. Producer<String> producer = client.newProducer(Schema.STRING)
    2.         .topic("my-topic")
    3.         .enableBatch(false)
    4.         .create();
    5. // 3 messages with "key-1", 3 messages with "key-2", 2 messages with "key-3" and 2 messages with "key-4"
    6. producer.newMessage().key("key-1").value("message-1-1").send();
    7. producer.newMessage().key("key-1").value("message-1-2").send();
    8. producer.newMessage().key("key-1").value("message-1-3").send();
    9. producer.newMessage().key("key-2").value("message-2-1").send();
    10. producer.newMessage().key("key-2").value("message-2-2").send();
    11. producer.newMessage().key("key-2").value("message-2-3").send();
    12. producer.newMessage().key("key-3").value("message-3-1").send();
    13. producer.newMessage().key("key-3").value("message-3-2").send();
    14. producer.newMessage().key("key-4").value("message-4-1").send();
    15. producer.newMessage().key("key-4").value("message-4-2").send();

    Exclusive

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

    1. Consumer consumer = client.newConsumer()
    2.         .topic("my-topic")
    3.         .subscriptionName("my-subscription")
    4.         .subscriptionType(SubscriptionType.Exclusive)
    5.         .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 mode.

    1. Consumer consumer1 = client.newConsumer()
    2.         .topic("my-topic")
    3.         .subscriptionName("my-subscription")
    4.         .subscriptionType(SubscriptionType.Failover)
    5.         .subscribe()
    6. Consumer consumer2 = client.newConsumer()
    7.         .topic("my-topic")
    8.         .subscriptionName("my-subscription")
    9.         .subscriptionType(SubscriptionType.Failover)
    10.         .subscribe()
    11. //conumser1 is the active consumer, consumer2 is the standby consumer.
    12. //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 becomes active consumer.

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

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

    consumer2 will receive:

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

    Shared

    Create new consumers and subscribe with Shared subscription mode:

    1. Consumer consumer1 = client.newConsumer()
    2.         .topic("my-topic")
    3.         .subscriptionName("my-subscription")
    4.         .subscriptionType(SubscriptionType.Shared)
    5.         .subscribe()
    7. Consumer consumer2 = client.newConsumer()
    8.         .topic("my-topic")
    9.         .subscriptionName("my-subscription")
    10.         .subscriptionType(SubscriptionType.Shared)
    11.         .subscribe()
    12. //Both consumer1 and consumer 2 is active consumers.

    In shared subscription mode, multiple consumers can attach to the same subscription and message are delivered in a round robin distribution across consumers.

    If a broker dispatches only one message at a time, consumer1 will receive:

    consumer 2 will receive:

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

    Shared subscription is different from Exclusive and Failover subscription modes. Shared subscription has better flexibility, but cannot provide order guarantee.

    Key_shared

    This is a new subscription mode since 2.4.0 release, create new consumers and subscribe with Key_Shared subscription mode:

    1. Consumer consumer1 = client.newConsumer()
    2.         .topic("my-topic")
    3.         .subscriptionName("my-subscription")
    4.         .subscriptionType(SubscriptionType.Key_Shared)
    5.         .subscribe()
    7. Consumer consumer2 = client.newConsumer()
    8.         .topic("my-topic")
    10.         .subscriptionType(SubscriptionType.Key_Shared)
    11.         .subscribe()
    12. //Both consumer1 and consumer2 are active consumers.

    Key_Shared subscription is like Shared subscription, all consumers can attach to the same subscription. But it is different from Key_Shared subscription, 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 will receive:

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

    consumer 2 will receive:

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

    Note:

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

    Reader interface

    With the , Pulsar clients can "manually position" themselves within a topic, reading all messages from a specified message onward. The Pulsar API for Java enables you to create Reader objects by specifying a topic, a , and ReaderConfiguration.

    Here's an example:

    1. ReaderConfiguration conf = new ReaderConfiguration();
    2. byte[] msgIdBytes = // Some message ID byte array
    3. MessageId id = MessageId.fromByteArray(msgIdBytes);
    4. Reader reader = pulsarClient.newReader()
    5.         .topic(topic)
    6.         .startMessageId(id)
    7.         .create();
    9. while (true) {
    10.     Message message = reader.readNext();
    11.     // Process message
    12. }

    In the example above, a Reader object is instantiated for a specific topic and message (by ID); the reader then iterates over each message in the topic after the message 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.

    In Pulsar, all message data consists of byte arrays "under the hood." 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[]. Here's an example:

    1. Producer<byte[]> producer = client.newProducer()
    2.         .topic(topic)
    3.         .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'll need to specify a schema that informs Pulsar which data type will be transmitted over the .

    Schema example

    Let's say that you have a SensorReading class that you'd like to transmit over a Pulsar topic:

    1. public class SensorReading {
    2.     public float temperature;
    4.     public SensorReading(float temperature) {
    5.         this.temperature = temperature;
    6.     }
    8.     // A no-arg constructor is required
    9.     public SensorReading() {
    10.     }
    12.     public float getTemperature() {
    13.         return temperature;
    14.     }
    16.     public void setTemperature(float temperature) {
    17.         this.temperature = temperature;
    18.     }
    19. }

    You could then create a Producer<SensorReading> (or Consumer<SensorReading>) like so:

    1. Producer<SensorReading> producer = client.newProducer(JSONSchema.of(SensorReading.class))
    2.         .topic("sensor-readings")
    3.         .create();

    The following schema formats are currently available for Java:

    • No schema or the byte array schema (which can be applied using Schema.BYTES):
    1. Producer<byte[]> bytesProducer = client.newProducer(Schema.BYTES)
    2.       .topic("some-raw-bytes-topic")
    3.       .create();

    Or, equivalently:

    1. Producer<byte[]> bytesProducer = client.newProducer()
    2.       .topic("some-raw-bytes-topic")
    3.       .create();
    • String for normal UTF-8-encoded string data. This schema can be applied using Schema.STRING:
    1. Producer<String> stringProducer = client.newProducer(Schema.STRING)
    2.       .topic("some-string-topic")
    3.       .create();
    • JSON schemas can be created for POJOs using the JSONSchema class. Here's an example:
    1. Schema<MyPojo> pojoSchema = JSONSchema.of(MyPojo.class);
    2. Producer<MyPojo> pojoProducer = client.newProducer(pojoSchema)
    3.       .topic("some-pojo-topic")
    4.       .create();

    Authentication

    Pulsar currently supports two authentication schemes: TLS and . The Pulsar Java client can be used with both.

    TLS Authentication

    To use , you need to set TLS to true using the setUseTls method, point your Pulsar client to a TLS cert path, and provide paths to cert and key files.

    Here's an example configuration:

    1. Map<String, String> authParams = new HashMap<>();
    2. authParams.put("tlsCertFile", "/path/to/client-cert.pem");
    3. authParams.put("tlsKeyFile", "/path/to/client-key.pem");
    5. Authentication tlsAuth = AuthenticationFactory
    6.         .create(AuthenticationTls.class.getName(), authParams);
    8. PulsarClient client = PulsarClient.builder()
    9.         .serviceUrl("pulsar+ssl://my-broker.com:6651")
    10.         .enableTls(true)
    11.         .tlsTrustCertsFilePath("/path/to/cacert.pem")
    12.         .authentication(tlsAuth)
    13.         .build();

    To use Athenz as an authentication provider, you need to and provide values for four parameters in a hash:

    • tenantDomain
    • tenantService
    • providerDomain
    • privateKeyYou can also set an optional keyId. Here's an example configuration: