Creating Apache Pulsar™ Producers, Consumers and Readers in Java

Difficulty: Beginner

Estimated Time: 20 minutes

In this scenario, you will:

Create a Pulsar producer client in Java
Create a Pulsar consumer client in Java
Create a Pulsar reader client in Java
Use multiple consumers with a shared subscription on a topic to create a work queue ("competing consumers" pattern)
Use readers to replay messages from a topic starting at the earliest available message

Check out our datastax.com/dev site, where you can find many more resources to help you succeed with Apache Cassandra™.

In this scenario, you learned how to:

Create a Pulsar producer client in Java
Create a Pulsar consumer client in Java
Create a Pulsar reader client in Java
Use multiple consumers with a shared subscription on a topic to create a work queue ("competing consumers" pattern)
Use readers to replay messages from a topic starting at the earliest available message

Your environment is currently being packaged as a Docker container and the download will begin shortly. To run the image locally, once Docker has been installed, use the commands

cat scrapbook_datastax_pulsar/java-client_container.tar | docker load

docker run -it /datastax_pulsar/java-client:

Restart Scenario

Next Scenario

Step 1 of 10

Explore How Java Projects are Configured

In this scenario, Apache Pulsar™ Producer, Consumer and Reader applications are created using Apache Maven™. The three projects have identical configurations.

Start by exploring a project configuration in one of the pom.xml files below (click to open):

my-pulsar-producer/pom.xml

my-pulsar-consumer/pom.xml

my-pulsar-reader/pom.xml

Most importantly, verify the properties and dependencies:

<!-- in your <properties> block -->
<pulsar.version>2.6.2</pulsar.version>

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

Do Not Skip This Step

Initialize all the Terminals in this order to avoid a bug in Katacoda:

echo "Initialize Terminal"

echo "Initialize Admin Terminal"

echo "Initialize Producer Terminal"

echo "Initialize Consumer 1 Terminal"

echo "Initialize Consumer 2 Terminal"

echo "Initialize Reader Terminal"

Create Producer App

1. Open file MyPulsarProducer.java:

my-pulsar-producer/src/main/java/com/example/pulsar/MyPulsarProducer.java

2. Import PulsarClient, Producer, Schema and PulsarClientException:

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.PulsarClientException;

3. Instantiate a PulsarClient object:

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

4. Create a Producer for topic persistent://public/default/my-topic:

      Producer producer = 
        client.newProducer(Schema.STRING)
              .topic("persistent://public/default/my-topic")
              .create();

5. Send 120 messages to the topic at the rate of 1 message per second:

      for (int i = 1 ; i <= 120 ; i++) {
        producer.send( "My message " + i );
        java.util.concurrent.TimeUnit.SECONDS.sleep(1);
      }

6. Close the producer and client, and release all the resources:

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

7. Declare exceptions:

throws PulsarClientException, java.lang.InterruptedException

8. Build the producer application:

mvn -f my-pulsar-producer/pom.xml clean compile assembly:single

Create Consumer App

1. Open file MyPulsarConsumer.java:

my-pulsar-consumer/src/main/java/com/example/pulsar/MyPulsarConsumer.java

2. Import PulsarClient, Consumer, Message, SubscriptionType and PulsarClientException:

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.PulsarClientException;

3. Instantiate a PulsarClient object:

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

4. Create a Consumer and subscribe to topic persistent://public/default/my-topic:

      Consumer consumer = 
        client.newConsumer()
              .topic("persistent://public/default/my-topic")
              .subscriptionName("my-subscription")
              .subscriptionType(SubscriptionType.Shared)
              .subscribe();

5. Receive 60 messages:

      for (int i = 1 ; i <= 60 ; i++) {
        Message msg = consumer.receive();

        try {
          System.out.println("RECEIVED: " + new String(msg.getData()));
          // Acknowledge the message
          consumer.acknowledge(msg);
        } catch (Exception e) {
          // Message failed to process, redeliver later
          consumer.negativeAcknowledge(msg);
        }
      }

6. Close the consumer and client, and release all the resources:

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

7. Declare exceptions:

throws PulsarClientException

8. Build the consumer application:

mvn -f my-pulsar-consumer/pom.xml clean compile assembly:single

Create Reader App

1. Open file MyPulsarReader.java:

my-pulsar-reader/src/main/java/com/example/pulsar/MyPulsarReader.java

2. Import PulsarClient, Reader, Message, MessageId and PulsarClientException:

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;

3. Instantiate a PulsarClient object:

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

4. Create a Reader on topic persistent://public/default/my-topic starting at the earliest available message:

      Reader reader = 
        client.newReader()
              .topic("persistent://public/default/my-topic")
              .startMessageId(MessageId.earliest)
              .create();

5. Read 120 messages:

      for (int i = 1 ; i <= 120 ; i++) {
        Message msg = reader.readNext();
        if(msg != null) System.out.println("RECEIVED: " + new String(msg.getData()));
      }

6. Close the reader and client, and release all the resources:

      reader.close();
      client.close();

7. Declare exceptions:

throws PulsarClientException, java.io.IOException

8. Build the reader application:

mvn -f my-pulsar-reader/pom.xml clean compile assembly:single

Start Apache Pulsar™

Start a local cluster using the pulsar command:

pulsar standalone

If Apache Pulsar™ starts successfully, there will be a number of INFO-level log messages displayed.

Create New Topic

Create topic my-topic in the public/default namespace:

pulsar-admin topics create persistent://public/default/my-topic

Display the topics in the public/default namespace:

pulsar-admin topics list public/default

Set Retention Policy

Configure retention to be unlimited both in size and time for the public/default namespace:

pulsar-admin namespaces set-retention public/default \
 --size -1 \
 --time -1

Create Work Queue with Two Consumers and One Producer

Start Consumer in Consumer 1 Terminal:

java -cp my-pulsar-consumer/target/my-pulsar-consumer-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.pulsar.MyPulsarConsumer

Start Consumer in Consumer 2 Terminal:

java -cp my-pulsar-consumer/target/my-pulsar-consumer-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.pulsar.MyPulsarConsumer

Start Producer in Producer Terminal:

java -cp my-pulsar-producer/target/my-pulsar-producer-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.pulsar.MyPulsarProducer

Switch between the consumer terminals to observe how the messages are being processed.

Replay Messages Using Reader

At this point, the consumers might have already processed and acknowledged all the messages published by the producer. However, the messages are still retained according to the retention policy and can be replayed using the reader.

Start Reader in Reader Terminal:

java -cp my-pulsar-reader/target/my-pulsar-reader-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.pulsar.MyPulsarReader

Replay the messages one more time:

java -cp my-pulsar-reader/target/my-pulsar-reader-1.0-SNAPSHOT-jar-with-dependencies.jar com.example.pulsar.MyPulsarReader

Terminal

Admin Terminal

Producer Terminal

Consumer 1 Terminal

Consumer 2 Terminal

Reader Terminal

cd <directory>	Change directory
ls	List directory
echo 'contents' > <file>	Write contents to a file
cat <file>	Output contents of file

i	In Command Mode	Change into Insert Mode, you can now insert and edit text in the file
esc	In Insert Mode	Change into Command Mode, you can now execute commands
:wq	In Command Mode	Save and Exit

Welcome!

Creating Apache Pulsar™ Producers, Consumers and Readers in Java

Congratulations!

You've completed the scenario!

Scenario Rating

Steps