Using the Event Hub Java SDK

Configuring Environmental Variables and VCAP_SERVICES

EventHubConfiguration eventHubConfiguration = new EventHubConfiguration.Builder()
    .fromEnvironmentVariables().build();

EventHubConfiguration eventHubConfiguration = new EventHubConfiguration.Builder()
    .fromEnvironmentVariables("<event-hub-instance-service-name>", "<uaa-instance-service-name>")
    .build();

The CLIENT_ID and CLIENT_SECRET can be set in tandem with fromEnvironmentVariables(…)

// If the clientID and clientSecret are not in environment variables
EventHubConfiguration eventHubConfiguration = new EventHubConfiguration.Builder()
    .clientID("id")
    .clientSecret("secret")
    .fromEnvironmentVariables("event-hub-instance-service-name", "uaa-instance-service-name")
    .build();

Using Event Hub Without the Predix UAA Instance

The following example does not require an UAA instance to be bound. It pulls only the Event Hub information from VCAP_SERVICES leaving all OAuth2 details as null. This is useful if the OAuth2 or authentication provider is not in Predix. However, setAuthToken() must be called to set a token to connect to Event Hub and automaticTokenRenew() must be false.

EventHubConfiguration eventHubConfiguration = new EventHubConfiguration.Builder()
    .fromEnvironmentVariables("event-hub-instance-service-name")
    .automaticTokenRenew(false)
    .build();

The following pulls only the Event Hub information from VCAP_SERVICES leaving all OAuth2 details as null, so a bound UAA service instance is not required. This is useful if the OAuth2 or authentication provider is not in Predix. However, setAuthToken() must be called to set a token to connect to Event Hub and automaticTokenRenew() must be false.

EventHubConfiguration eventHubConfiguration =newEventHubConfiguration.Builder()
    .fromEnvironmentVariables("event-hub-instance-service-name")
    .automaticTokenRenew(false)
    .build();

Connecting to the Event Hub

Client eventHubClient  = new Client(EventHubConfiguration configuration);

Example

EventHubConfiguration configuration =newEventHubConfiguration.Builder()
          ...
          .publishConfiguration(newPublishAsyncConfiguration.Builder()
            .maxAddedMessages(500)
            .build())
          ...
          .build();

Async Configuration

The async publishing mode uses the PublishAsyncConfiguration configuration. The async parameters determine the type of and rate of acknowledgements sent from Event Hub.

Example

EventHubConfiguration configuration =newEventHubConfiguration.Builder()
          ...
          .publishConfiguration(newPublishAsyncConfiguration.Builder()
            .ackType(PublishAsyncConfiguration.AcknowledgementOptions.NONE)
            .cacheAckIntervalMillis(200)
            .build())
          ...
          .build();

Sync Configuration

The sync publishing mode uses the PublishSyncConfiguration configuration. When publishing, the publisher will block and wait for the acks and nacks. However, the total throughput may be less than async.

Example

EventHubConfiguration configuration =newEventHubConfiguration.Builder()
          ...
          .publishConfiguration(newPublishSyncConfiguration.Builder()
              .timeout(2000)
              .build())
          ...
          .build();

Shared

The field maxAddedMessages can be used in both sync and async publish modes:

ventHubConfiguration configuration = new EventHubConfiguration.Builder()
          ...
          .publishConfiguration(new PublishAsyncConfiguration.Builder()
            .maxAddedMessages(500)
            .build())
          ...
          .build();

          EventHubConfiguration configuration = new EventHubConfiguration.Builder()
          ...
          .publishConfiguration(new PublishSyncConfiguration.Builder()
            .maxAddedMessages(500)
            .build())
          ...
          .build();

Sending Messages

Messages are added to the client before sending. A message contents contains the following three fields:

Message.newBuilder().setId("id").setZoneId("zoneId")
        .setBody(ByteString.copyFromUtf8("body"));

eventHubClient.addMessage(String id, String body, Map<String, String> tags);
eventHubClient.addMessage(Message newMessage);
eventHubClient.addMessages(Messages newMessages);

eventHubClient.addMessages(newMessages).addMessage("1","body",null);

Receiving Acknowledgements

In sync mode, the send methods returns List<Ack> which contains the acks for the messages sent. Async mode does not return anything because the acks are received in a callback.

eventHubClient.flush(); // if async mode
List<Ack> messages = eventHubClient.flush(); // if sync mode

eventHubClient.send(Message message); // if async mode
List<Ack> message = eventHubClient.send(Message message); // if sync mode

public interface PublishCallback {
    void onAck(List<Ack> acks);
    void onFailure(Throwable throwable);
  }

acks and nacks are returned through the callback's onAck() method. Connection closure or errors are returned through the callback's onFailure() method.

eventHubClient.registerPublishCallback(Client.PublishCallback callback);

To reopen the stream after, call either forceRenewToken() or setAuthToken() which will use the new token and reconnect the stream.

Example

EventHubConfiguration configuration = new EventHubConfiguration.Builder()
          ...
          .subscribeConfiguration(new SubscribeConfiguration.Builder()
            .subscriberName("name")
            .retryIntervalSeconds(30)
            .subscribeRecency(EventHubConfiguration.SubscribeRecency.NEWEST)
            .build())
          ...
          .build();

        Client eventHubClient = new Client(eventHubConfiguration);

Single Subscription

After a Client is created, then Subscribe or subscribeWithAcks can be called by providing a callback to receive the messages. The callback must implement the following interface.

 public interface SubscribeCallback {
    void onMessage(Message message); //Called when new message received
    void onFailure(Throwable throwable); //Called with stream or connection issue
  }

  eventHubClient.subscribe(Client.SubscribeCallback callback);

A received message triggers the onMessage() method. Similarly to the publish callback, onFailure() is called if there is a stream or connection issue.

Messages can be acknowledged as follows:

eventHubClient.sendAck(Message message);
eventHubClient.sendAcks(Messages messages);

unsubscribe() can be called to close the subscription stream, eventHubClient.unsubscribe();.

Either subscribe or subscribeWithAck can be used to recreate the subscription. Note that creating subscribe and subscribeWithAck subscriptions is not supported.

Multiple Subscriptions

Closing Connections

eventHubClient.shutdown();

Automatic Reconnection

By default, the SDK will attempt to reconnect to the Event Hub upon disconnection. It employs a back off with increasing delays which recycles based on the error. The following are some of the errors and how they are handled.

UNAVAILABLE

UNAUTHENTICATED

UNKNOWN

Client

Renewing the Token

Given OAuth2 parameters are set, the default configuration will renew and cache the token as needed. However, there are methods for finer control of the token. automaticTokenRenew is an optional configuration which has a default of True. Setting to False prevents the automatic retrieval of tokens and the use of the SDK to get a token

EventHubConfiguration configuration = new EventHubConfiguration.Builder()
    ...
    .automaticTokenRenew(Boolean bool)
    .build()

Use the following method to update the tokens manually.

eventHubClient.setAuthToken(String token);

Both forceRenewToken() and setAuthToken() restart the connection for the client to ensure the new tokens are active.

Requesting Specific Scopes

The SDK tries to requests all Event Hub scopes associated to the publish and subscribe by default. If it fails it will fall back to trying only either publish or subscribe scopes. However depending on the number of scopes the auth user has, the token length passed to Event Hub must not exceed 60,000 characters else the client may not be able to authenticate with Event Hub. Specifics scopes can also be passed into a EventHubConfiguration via authScopes(String scopes) where string is all the scopes comma separated. The following are the Event Hub scopes:

<zoneID>

<zoneID>

<zoneID>

Building the SDK from the Repository

After pulling the project, do a mvn clean compile which will automatically generate the protobuf classes.

This SDK uses Proton v3, which must also be used when compiling the proton file manually. Link to the Protobuf compiler protoc (if generating manually):

Common Issues

In order to use SSL over HTTP2, BoringSSL needs to be used. The following dependency needs to be listed first in the pom.xml of the project using the Java SDK.

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-tcnative-boringssl-static</artifactId>
    <version>1.1.33.Fork23</version>
</dependency>
<repositories>
  <repository>
    <id>netty-repo-2</id>
    <url>http://repo1.maven.org/maven2/</url>
  </repository>
</repositories>