srnagar

(all names in this doc are not finalized yet)

EPH APIs

/**
 * This is the starting point for event processor. To create an instance of this, use the
 * {@link EventProcessorBuilder}
 */
public class EventProcessor {
    

srnagar

Load balancing for Event Processor is based on the number of Event Hub partitions and number of Event Processor instances available. This assumes all Event Processor instances are identical and events from all partitions require same processing and the events are evenly distributed among partitions.

Terminology

Store - refers to any of the underlying mechanisms used to persist partition owernship + checkpoint information. It could be in-memory (for testing purposes), Azure blobs, Azure Tables or any other storage that provides strong consistency. Strong consistency is a requirement for this to work.

Partition id - An int (as string) value ranging from 0 to P-1. Currently, P is immutable once an Event Hub is setup. However, this is going to change in the future.

Owner id - Unique string identifier for an instance of Event Processor.

srnagar

Alternate solution to offset provider

In track 1 library, Event Processor Host (EPH) used an offset provider to bootstrap an application for the first time. When an EPH starts up and claims ownership of a partition, it needs to determine the position (sequence number/offset) to start processing events from. If a checkpoint is available, the EPH will start processing the events from the checkpointed sequence number. If it's not available (i.e the application has never used EPH before or the storage is new/migrated and all checkpoints are lost), then EPH will fallback to using user provided implementation of an offset provider that has the ability to return a sequence number to start processing from for a given partition id.

Given that the offset provider is used occasionally and increases the complexity of EPH interface and implementation, in Track 2 library, the offset provider can be deprecated and Event Processor (EPH equivalent in Track 2) can be simplified. If/when applications want to provide a cust

srnagar

Event Processor API proposal

Proposal 1

Key changes:

• Creating an instance Event Processor will have similar constructor overloads that Event Hub Client has with two additional parameters:
  • Consumer group name

srnagar

Proposal to have separate partition ownership and checkpoint model class

Currently, PartitionOwnership class contains both ownership and checkpoint information.

Advantages

• Decouples checkpointing and load balancing
• Allows users to opt-in to either checkpointing or load balancing or both
• Each of them can evolve separately without breaking functionality of the other - for e.g. adding new information to checkpointing (custom checkpoints the service team talked about) doesn't impact load balancing (ownership) APIs.
• Checkpoints are customer data while partition ownership is event processor data. Customer may want to see/analyse checkpoint information and partition ownership data can be hidden from the customer.

srnagar

    // ---------------------Receive Operations----------------------------------
    /* Receive events from all partitions with no load balancing or checkpointing*/
    private static void receiveAll() {
        EventHubConsumerClient consumerClient = new EventHubClientBuilder()
            .connectionString("")
            .consumerGroup("")
            .buildConsumerClient();

	// Have to combine event data and partition context to return a Flux

srnagar

Publish events

Creating a producer client

// with connection string
EventHubProducerClient producerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .buildProducer();
// with connection string and event hub name                                        
EventHubProducerClient producerClient = new EventHubClientBuilder()

srnagar

    // ---------------------Send Operations----------------------------------
    /* Send events without specifying partition */
    private static void send(List<EventData> eventDataList) {
        EventHubProducerClient producerClient = new EventHubClientBuilder()
            .connectionString("")
            .buildProducerClient();
        producerClient.send(eventDataList);
    }

srnagar

Querying Event Hub properties

public void queryEventHubProperties() {
    EventHubProducerClient client = new EventHubClientBuilder()
        .connectionString("connection-string")
        .buildProducerClient();

    EventHubProperties eventHubProperties = client.getEventHubProperties();
    System.out.println("Name = " + eventHubProperties.getName());

srnagar

Error Handling

Error	Where?	Action	Notes
Error from listing ownership	Top level
Error from claiming ownership	Top level
Error from listing checkpoints	Top level
Error getting Event Hub properties	Top level
Error receiving events from a partition	Partition level
Error updating checkpoints	Partition level