com.azure.messaging.eventhubs

Class EventProcessorClientBuilder

java.lang.Object

com.azure.messaging.eventhubs.EventProcessorClientBuilder

public class EventProcessorClientBuilder
extends Object

This class provides a fluent builder API to help aid the configuration and instantiation of the EventProcessorClient. Calling EventProcessorClientBuilder.buildEventProcessorClient() constructs a new instance of EventProcessorClient.

To create an instance of EventProcessorClient that processes events with user-provided callback, configure the following fields:

• Consumer group name.
• CheckpointStore - An implementation of CheckpointStore that stores checkpoint and partition ownership information to enable load balancing.
• EventProcessorClientBuilder.processEvent(Consumer) - A callback that processes events received from the Event Hub.
• Credentials - Credentials are required to perform operations against Azure Event Hubs. They can be set by using one of the following methods:
  • connectionString(String) with a connection string to a specific Event Hub.
  • connectionString(String, String) with an Event Hub namespace connection string and the Event Hub name.
  • credential(String, String, TokenCredential) with the fully qualified namespace, Event Hub name, and a set of credentials authorized to use the Event Hub.

Creating an EventProcessorClient

public EventProcessorClient createEventProcessor() {
    String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};"
        + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}";

    EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
        .consumerGroup("consumer-group")
        .checkpointStore(new InMemoryCheckpointStore())
        .processEvent(eventContext -> {
            System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId()
                + "and sequence number of event = " + eventContext.getEventData().getSequenceNumber());
        })
        .processError(errorContext -> {
            System.out.printf("Error occurred in partition processor for partition {}, {}",
                errorContext.getPartitionContext().getPartitionId(),
                errorContext.getThrowable());
        })
        .connectionString(connectionString)
        .buildEventProcessorClient();
    return eventProcessorClient;
}

See Also:

EventProcessorClient, EventHubConsumerClient, EventHubConsumerAsyncClient

Constructor Summary

Constructors

Constructor and Description
EventProcessorClientBuilder() Creates a new instance of EventProcessorClientBuilder.

Method Summary

Modifier and Type	Method and Description
EventProcessorClient	buildEventProcessorClient() This will create a new EventProcessorClient configured with the options set in this builder.
EventProcessorClientBuilder	checkpointStore(CheckpointStore checkpointStore) Sets the CheckpointStore the EventProcessorClient will use for storing partition ownership and checkpoint information.
EventProcessorClientBuilder	configuration(Configuration configuration) Sets the configuration store that is used during construction of the service client.
EventProcessorClientBuilder	connectionString(String connectionString) Sets the credential information given a connection string to the Event Hub instance.
EventProcessorClientBuilder	connectionString(String connectionString, String eventHubName) Sets the credential information given a connection string to the Event Hubs namespace and name to a specific Event Hub instance.
EventProcessorClientBuilder	consumerGroup(String consumerGroup) Sets the consumer group name from which the EventProcessorClient should consume events.
EventProcessorClientBuilder	credential(String fullyQualifiedNamespace, String eventHubName, TokenCredential credential) Sets the credential information for which Event Hub instance to connect to, and how to authorize against it.
EventProcessorClientBuilder	processError(Consumer<ErrorContext> processError) The function that is called when an error occurs while processing events.
EventProcessorClientBuilder	processEvent(Consumer<EventContext> processEvent) The function that is called for each event received by this EventProcessorClient.
EventProcessorClientBuilder	processPartitionClose(Consumer<CloseContext> closePartition) The function that is called when a processing for a partition stops.
EventProcessorClientBuilder	processPartitionInitialization(Consumer<InitializationContext> initializePartition) The function that is called before processing starts for a partition.
EventProcessorClientBuilder	proxyOptions(ProxyOptions proxyOptions) Sets the proxy configuration to use for EventHubAsyncClient.
EventProcessorClientBuilder	retry(AmqpRetryOptions retryOptions) Sets the retry policy for EventHubAsyncClient.
EventProcessorClientBuilder	trackLastEnqueuedEventProperties(boolean trackLastEnqueuedEventProperties) Sets whether or not the event processor should request information on the last enqueued event on its associated partition, and track that information as events are received.
EventProcessorClientBuilder	transportType(AmqpTransportType transport) Sets the transport type by which all the communication with Azure Event Hubs occurs.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

EventProcessorClientBuilder

public EventProcessorClientBuilder()

Creates a new instance of EventProcessorClientBuilder.

Method Detail

connectionString

public EventProcessorClientBuilder connectionString(String connectionString)

Sets the credential information given a connection string to the Event Hub instance.

If the connection string is copied from the Event Hubs namespace, it will likely not contain the name to the desired Event Hub, which is needed. In this case, the name can be added manually by adding "EntityPath=EVENT_HUB_NAME" to the end of the connection string. For example, "EntityPath=telemetry-hub".

If you have defined a shared access policy directly on the Event Hub itself, then copying the connection string from that Event Hub will result in a connection string that contains the name.

Parameters:

connectionString - The connection string to use for connecting to the Event Hub instance. It is expected that the Event Hub name and the shared access key properties are contained in this connection string.

Returns:

The updated EventProcessorClientBuilder object.

Throws:

NullPointerException - if connectionString is null.

IllegalArgumentException - if connectionString is empty. Or, the connectionString does not contain the "EntityPath" key, which is the name of the Event Hub instance.

AzureException - If the shared access signature token credential could not be created using the connection string.

connectionString

public EventProcessorClientBuilder connectionString(String connectionString,
                                                    String eventHubName)

Sets the credential information given a connection string to the Event Hubs namespace and name to a specific Event Hub instance.

Parameters:

connectionString - The connection string to use for connecting to the Event Hubs namespace; it is expected that the shared access key properties are contained in this connection string, but not the Event Hub name.

eventHubName - The name of the Event Hub to connect the client to.

Returns:

The updated EventProcessorClientBuilder object.

Throws:

NullPointerException - if connectionString or eventHubName is null.

IllegalArgumentException - if connectionString or eventHubName is an empty string. Or, if the connectionString contains the Event Hub name.

AzureException - If the shared access signature token credential could not be created using the connection string.

configuration

public EventProcessorClientBuilder configuration(Configuration configuration)

Sets the configuration store that is used during construction of the service client. If not specified, the default configuration store is used to configure the EventHubAsyncClient. Use Configuration.NONE to bypass using configuration settings during construction.

Parameters:

configuration - The configuration store used to configure the EventHubAsyncClient.

Returns:

The updated EventProcessorClientBuilder object.

credential

public EventProcessorClientBuilder credential(String fullyQualifiedNamespace,
                                              String eventHubName,
                                              TokenCredential credential)

Sets the credential information for which Event Hub instance to connect to, and how to authorize against it.

Parameters:

fullyQualifiedNamespace - The fully qualified name for the Event Hubs namespace. This is likely to be similar to "{your-namespace}.servicebus.windows.net".

eventHubName - The name of the Event Hub to connect the client to.

credential - The token credential to use for authorization. Access controls may be specified by the Event Hubs namespace or the requested Event Hub, depending on Azure configuration.

Returns:

The updated EventProcessorClientBuilder object.

Throws:

IllegalArgumentException - if fullyQualifiedNamespace or eventHubName is an empty string.

NullPointerException - if fullyQualifiedNamespace, eventHubName, credentials is null.

proxyOptions

public EventProcessorClientBuilder proxyOptions(ProxyOptions proxyOptions)

Sets the proxy configuration to use for EventHubAsyncClient. When a proxy is configured, AmqpTransportType.AMQP_WEB_SOCKETS must be used for the transport type.

Parameters:

proxyOptions - The proxy options to use.

Returns:

The updated EventProcessorClientBuilder object.

transportType

public EventProcessorClientBuilder transportType(AmqpTransportType transport)

Sets the transport type by which all the communication with Azure Event Hubs occurs. Default value is AmqpTransportType.AMQP.

Parameters:

transport - The transport type to use.

Returns:

The updated EventProcessorClientBuilder object.

retry

public EventProcessorClientBuilder retry(AmqpRetryOptions retryOptions)

Sets the retry policy for EventHubAsyncClient. If not specified, the default retry options are used.

Parameters:

retryOptions - The retry policy to use.

Returns:

The updated EventProcessorClientBuilder object.

consumerGroup

public EventProcessorClientBuilder consumerGroup(String consumerGroup)

Sets the consumer group name from which the EventProcessorClient should consume events.

Parameters:

consumerGroup - The consumer group name this EventProcessorClient should consume events.

Returns:

The updated EventProcessorClientBuilder instance.

Throws:

NullPointerException - if consumerGroup is null.

checkpointStore

public EventProcessorClientBuilder checkpointStore(CheckpointStore checkpointStore)

Sets the CheckpointStore the EventProcessorClient will use for storing partition ownership and checkpoint information.

Users can, optionally, provide their own implementation of CheckpointStore which will store ownership and checkpoint information.

Parameters:

checkpointStore - Implementation of CheckpointStore.

Returns:

The updated EventProcessorClientBuilder instance.

Throws:

NullPointerException - if checkpointStore is null.

processEvent

public EventProcessorClientBuilder processEvent(Consumer<EventContext> processEvent)

The function that is called for each event received by this EventProcessorClient. The input contains the partition context and the event data.

Parameters:

processEvent - The callback that's called when an event is received by this EventProcessorClient.

Returns:

The updated EventProcessorClientBuilder instance.

Throws:

NullPointerException - if processEvent is null.

processError

public EventProcessorClientBuilder processError(Consumer<ErrorContext> processError)

The function that is called when an error occurs while processing events. The input contains the partition information where the error happened.

Parameters:

processError - The callback that's called when an error occurs while processing events.

Returns:

The updated EventProcessorClientBuilder instance.

processPartitionInitialization

public EventProcessorClientBuilder processPartitionInitialization(Consumer<InitializationContext> initializePartition)

The function that is called before processing starts for a partition. The input contains the partition information along with a default starting position for processing events that will be used in the case of a checkpoint unavailable in CheckpointStore. Users can update this position if a different starting position is preferred.

Parameters:

initializePartition - The callback that's called before processing starts for a partition

Returns:

The updated EventProcessorClientBuilder instance.

processPartitionClose

public EventProcessorClientBuilder processPartitionClose(Consumer<CloseContext> closePartition)

The function that is called when a processing for a partition stops. The input contains the partition information along with the reason for stopping the event processing for this partition.

Parameters:

closePartition - The callback that's called after processing for a partition stops.

Returns:

The updated EventProcessorClientBuilder instance.

trackLastEnqueuedEventProperties

public EventProcessorClientBuilder trackLastEnqueuedEventProperties(boolean trackLastEnqueuedEventProperties)

Sets whether or not the event processor should request information on the last enqueued event on its associated partition, and track that information as events are received.

When information about the partition's last enqueued event is being tracked, each event received from the Event Hubs service will carry metadata about the partition that it otherwise would not. This results in a small amount of additional network bandwidth consumption that is generally a favorable trade-off when considered against periodically making requests for partition properties using the Event Hub client.

Parameters:

trackLastEnqueuedEventProperties - true if the resulting events will keep track of the last enqueued information for that partition; false otherwise.

Returns:

The updated EventProcessorClientBuilder instance.

buildEventProcessorClient

public EventProcessorClient buildEventProcessorClient()

This will create a new EventProcessorClient configured with the options set in this builder. Each call to this method will return a new instance of EventProcessorClient.

All partitions processed by this EventProcessorClient will start processing from earliest available event in the respective partitions.

Returns:

A new instance of EventProcessorClient.

Throws:

NullPointerException - if processEvent or processError or checkpointStore or consumerGroup is null.

IllegalArgumentException - if the credentials have not been set using either EventProcessorClientBuilder.connectionString(String) or EventProcessorClientBuilder.credential(String, String, TokenCredential). Or, if a proxy is specified but the transport type is not web sockets.