Connect to Pulsar cluster using Java client

​

Connect to a Pulsar cluster using a token

​

Prerequisites

• Get the service URL.
• Get a service account token.

​

Steps

1. Connect to the Pulsar cluster.
   package io.streamnative.examples.oauth2;
   
   import org.apache.pulsar.client.api.AuthenticationFactory;
   import org.apache.pulsar.client.api.PulsarClient;
   
   public class ConnectByToken {
     public static void main(String[] args) throws Exception {
       PulsarClient client = PulsarClient.builder()
         .serviceUrl("SERVICE_URL")
         .authentication(AuthenticationFactory.token("AUTH_PARAMS"))
         .build();
   
       client.close();
     }
   }
   
   Set the SERVICE_URL and AUTH_PARAMS parameters based on the descriptions in the prepare to connect to a Pulsar cluster user guide.
2. Create a Java consumer and use the Java consumer to consume messages.
   public class SampleConsumer {
     public static void main(String[] args) throws Exception {
       String serviceUrl = "SERVICE_URL";
       String topic = "persistent://public/default/topic-1";
       String token = "YOUR_TOKEN";
   
       PulsarClient client = PulsarClient.builder()
         .serviceUrl(serviceUrl)
         .authentication(AuthenticationFactory.token(token))
         .build();
   
       Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)
         .topic(topic)
         .subscriptionName("sub")
         .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
         .subscribe();
   
   
       for (int i = 0; i < 10; i++) {
         Message<byte[]> msg = consumer.receive();
         consumer.acknowledge(msg);
         System.out.println("Receive message " + new String(msg.getData()));
       }
   
       consumer.close();
       client.close();
     }
   }
   
3. Create a Java producer and use the Java producer to produce messages.
   public class SampleProducer {
     public static void main(String[] args) throws Exception {
       String serviceUrl = "SERVICE_URL";
       String topic = "persistent://public/default/topic-1";
       String token = "YOUR_TOKEN";
   
       PulsarClient client = PulsarClient.builder()
         .serviceUrl(serviceUrl)
         .authentication(AuthenticationFactory.token(token))
         .build();
   
       Producer<byte[]> producer = client.newProducer().topic(token).create();
   
   
   
       for (int i = 0; i < 10; i++) {
         String message = "my-message-" + i;
         MessageId msgID = producer.send(message.getBytes());
         System.out.println("Publish " + "my-message-" + i + " and message ID " + msgID);
       }
   
       producer.close();
       client.close();
     }
   }
   

​

Connect to a Pulsar cluster using an OAuth2 credential file

1. Generate the App credentials by following similar instructions in configure OAuth2 authentication.
2. Save the App credentials into an OAuth2 credential file.
3. Connect to your Pulsar cluster through the OAuth2 credential file.
   package io.streamnative.examples.oauth2;
   
   import java.net.URL;
   import org.apache.pulsar.client.api.PulsarClient;
   import org.apache.pulsar.client.impl.auth.oauth2.AuthenticationFactoryOAuth2;
   
   public class ConnectByOauth2 {
     public static void main(String[] args) throws Exception {
       String issuerUrl = "your-issuer-url";
       String credentialsUrl = "file:///path/to/private-key-file.json";
       String audience = "your-audience";
   
       PulsarClient client = PulsarClient.builder()
         .serviceUrl("pulsar-service-url")
         .authentication(
           AuthenticationFactoryOAuth2.clientCredentials(new URL(issuerUrl), new URL(credentialsUrl), audience))
         .build();
   
       client.close();
     }
   }
   
   • issuerUrl: the URL of your OAuth2 identity provider.
   • credentialsUrl: the path to your OAuth2 credential file.
   • audience: the audience of your Pulsar cluster.
   • serviceUrl: the URL of your Pulsar cluster.
4. Create a Java consumer and use the Java consumer to consume messages. You can create and configure the Java consumer to consume messages using the OAuth2 credential file. Or, you can define a Base64 String when creating the Java consumer. Then, the consumer can extract the client ID and client Secret from the OAuth2 credential file and use them to consume messages.
   • Using the OAuth2 credential file
     public class SampleConsumer {
       public static void main(String[] args) throws Exception {
         JCommanderPulsar jct = new JCommanderPulsar();
         JCommander jCommander = new JCommander(jct, args);
         if (jct.help) {
           jCommander.usage();
           return;
         }
     
         String topic = "your-topic";
     
         PulsarClient client = PulsarClient.builder()
           .serviceUrl(jct.serviceUrl)
           .authentication(
                   AuthenticationFactoryOAuth2.clientCredentials(new URL(jct.issuerUrl), new URL(jct.credentialsUrl), jct.audience))
           .build();
     
         Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)
           .topic(topic)
           .subscriptionName("your-sub-name")
           .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
           .subscribe();
     
         for (int i = 0; i < 10; i++) {
           Message<byte[]> msg = consumer.receive();
           consumer.acknowledge(msg);
           System.out.println("Receive message " + new String(msg.getData()));
         }
     
         consumer.close();
         client.close();
       }
     }
     
   • Using the OAuth2 client ID and client Secret
     package io.streamnative.examples.oauth2;
     
     import com.beust.jcommander.JCommander;
     import com.fasterxml.jackson.databind.ObjectMapper;
     import com.google.common.collect.Maps;
     import java.nio.charset.StandardCharsets;
     import java.util.Base64;
     import java.util.Map;
     import org.apache.pulsar.client.api.Consumer;
     import org.apache.pulsar.client.api.Message;
     import org.apache.pulsar.client.api.PulsarClient;
     import org.apache.pulsar.client.api.Schema;
     import org.apache.pulsar.client.api.SubscriptionInitialPosition;
     import org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2;
     
     public class SampleConsumerNoCredentialFile {
     
       public static void main(String[] args) throws Exception {
         JCommanderPulsar jct = new JCommanderPulsar();
         JCommander jCommander = new JCommander(jct, args);
         if (jct.help) {
           jCommander.usage();
           return;
         }
     
         String topic = "your-topic";
         AuthenticationOAuth2 authenticationOAuth2 = new AuthenticationOAuth2();
         ObjectMapper objectMapper = new ObjectMapper();
     
         Map<String, String> params = Maps.newHashMap();
         Map<String, String> data = Maps.newHashMap();
         data.put("client_id", jct.clientId);
         data.put("client_secret", jct.clientSecret);
         params.put("privateKey", "data:application/json;base64,"
           + new String(Base64.getEncoder().encode(
               objectMapper.writeValueAsString(data).getBytes(StandardCharsets.UTF_8))));
         params.put("issuerUrl", jct.issuerUrl);
         params.put("audience", jct.audience);
         params.put("scope", jct.scope);
         authenticationOAuth2.configure(objectMapper.writeValueAsString(params));
         PulsarClient client = PulsarClient.builder()
           .serviceUrl(jct.serviceUrl)
           .authentication(authenticationOAuth2)
           .build();
     
         Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)
           .topic(topic)
           .subscriptionName("your-sub-name")
           .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
           .subscribe();
     
         for (int i = 0; i < 10; i++) {
           Message<byte[]> msg = consumer.receive();
           consumer.acknowledge(msg);
           System.out.println("Receive message " + new String(msg.getData()));
         }
     
         consumer.close();
         client.close();
       }
     }
     
5. Create a Java producer and use the Java producer to produce messages. You can create and configure the Java producer to produce messages using the OAuth2 credential file. Or, you can define a Base64 String when creating the Java producer. Then, the producer can extract the client ID and client Secret from the OAuth2 credential file and use them to produce messages.
   • Using the OAuth2 credential file
     public class SampleProducer {
       public static void main(String[] args) throws Exception {
         JCommanderPulsar jct = new JCommanderPulsar();
         JCommander jCommander = new JCommander(jct, args);
         if (jct.help) {
           jCommander.usage();
           return;
         }
     
         String topic = "your-topic";
     
         PulsarClient client = PulsarClient.builder()
           .serviceUrl(jct.serviceUrl)
           .authentication(
             AuthenticationFactoryOAuth2.clientCredentials(new URL(jct.issuerUrl), new URL(jct.credentialsUrl), jct.audience))
           .build();
     
         ProducerBuilder<byte[]> producerBuilder = client.newProducer().topic(topic)
           .producerName("your-producer-name");
         Producer<byte[]> producer = producerBuilder.create();
     
         for (int i = 0; i < 10; i++) {
           String message = "my-message-" + i;
           MessageId msgID = producer.send(message.getBytes());
           System.out.println("Publish " + "my-message-" + i + " and message ID " + msgID);
         }
     
         producer.close();
         client.close();
       }
     }
     
   • Using the OAuth2 client ID and client Secret
     package io.streamnative.examples.oauth2;
     
     import com.beust.jcommander.JCommander;
     import com.fasterxml.jackson.databind.ObjectMapper;
     import com.google.common.collect.Maps;
     import java.nio.charset.StandardCharsets;
     import java.util.Base64;
     import java.util.Map;
     import org.apache.pulsar.client.api.MessageId;
     import org.apache.pulsar.client.api.Producer;
     import org.apache.pulsar.client.api.ProducerBuilder;
     import org.apache.pulsar.client.api.PulsarClient;
     import org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2;
     
     public class SampleProducerNoCredentialFile {
     
       public static void main(String[] args) throws Exception {
         JCommanderPulsar jct = new JCommanderPulsar();
         JCommander jCommander = new JCommander(jct, args);
         if (jct.help) {
           jCommander.usage();
           return;
         }
     
         String topic = "your-topic";
     
         AuthenticationOAuth2 authenticationOAuth2 = new AuthenticationOAuth2();
         ObjectMapper objectMapper = new ObjectMapper();
     
         Map<String, String> params = Maps.newHashMap();
         Map<String, String> data = Maps.newHashMap();
         data.put("client_id", jct.clientId);
         data.put("client_secret", jct.clientSecret);
         params.put("privateKey", "data:application/json;base64,"
           + new String(Base64.getEncoder().encode(
             objectMapper.writeValueAsString(data).getBytes(StandardCharsets.UTF_8))));
         params.put("issuerUrl", jct.issuerUrl);
         params.put("audience", jct.audience);
         params.put("scope", jct.scope);
         authenticationOAuth2.configure(objectMapper.writeValueAsString(params));
         PulsarClient client = PulsarClient.builder()
           .serviceUrl(jct.serviceUrl)
           .authentication(authenticationOAuth2)
           .build();
     
         ProducerBuilder<byte[]> producerBuilder = client.newProducer().topic(topic)
           .producerName("your-producer-name");
         Producer<byte[]> producer = producerBuilder.create();
     
         for (int i = 0; i < 10; i++) {
           String message = "my-message-" + i;
           MessageId msgID = producer.send(message.getBytes());
           System.out.println("Publish " + "my-message-" + i + " and message ID " + msgID);
         }
     
         producer.close();
         client.close();
       }
     }