skilld
All skills
microsoft avatar

/azure-eventhub-java

@e7ee58f
by microsoftmicrosoft/skills3.1k stars
351

Build real-time streaming applications with Azure Event Hubs SDK for Java. Use when implementing event streaming, high-throughput data ingestion, or building event-driven architectures.

Ask your Agent

Use this Skill: https://skilld.dev/gh/microsoft/skills/azure-eventhub-java

This session only. Nothing lands on disk.

referencesexamples.md

≈3.3k tokens on demand. Your agent reads this file only when SKILL.md points to it.

Azure Event Hubs Java SDK - Examples

Comprehensive code examples for the Azure Event Hubs SDK for Java.

Table of Contents

Maven Dependency

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-eventhubs</artifactId>
    <version>5.21.0</version>
</dependency>

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-identity</artifactId>
    <version>1.18.2</version>
</dependency>

<!-- For EventProcessorClient with blob checkpointing -->
<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
    <version>1.21.0</version>
</dependency>

EventHubProducerClient

Basic Producer with Azure Identity

import com.azure.messaging.eventhubs.*;
import com.azure.identity.DefaultAzureCredentialBuilder;

import java.util.Arrays;
import java.util.List;

import static java.nio.charset.StandardCharsets.UTF_8;

public class PublishEventsWithAzureIdentity {
    public static void main(String[] args) {
        List<EventData> telemetryEvents = Arrays.asList(
            new EventData("Roast beef".getBytes(UTF_8)),
            new EventData("Cheese".getBytes(UTF_8)),
            new EventData("Tofu".getBytes(UTF_8)),
            new EventData("Turkey".getBytes(UTF_8)));

        // Create a producer
        // "<<fully-qualified-namespace>>" = "{your-namespace}.servicebus.windows.net"
        EventHubProducerClient producer = new EventHubClientBuilder()
            .credential(
                "<<fully-qualified-namespace>>",
                "<<event-hub-name>>",
                new DefaultAzureCredentialBuilder().build())
            .buildProducerClient();

        // Create batch - Event Hubs auto-routes to available partitions
        EventDataBatch currentBatch = producer.createBatch();

        // Add events to batch, send when full
        for (EventData event : telemetryEvents) {
            if (currentBatch.tryAdd(event)) {
                continue;
            }

            // Batch is full - send and create new batch
            producer.send(currentBatch);
            currentBatch = producer.createBatch();

            if (!currentBatch.tryAdd(event)) {
                System.err.printf("Event is too large for an empty batch. Max size: %s.%n",
                    currentBatch.getMaxSizeInBytes());
            }
        }

        // Send remaining events
        producer.send(currentBatch);
        
        producer.close();
    }
}

Producer with Connection String

String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};"
    + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}";

EventHubProducerClient producer = new EventHubClientBuilder()
    .connectionString(connectionString)
    .buildProducerClient();

Send to Specific Partition

import com.azure.messaging.eventhubs.models.CreateBatchOptions;

// Route all events in batch to partition "0"
CreateBatchOptions options = new CreateBatchOptions().setPartitionId("0");
EventDataBatch batch = producer.createBatch(options);

batch.tryAdd(new EventData("Event for partition 0"));
producer.send(batch);

Send with Partition Key (Hash-based Routing)

import com.azure.messaging.eventhubs.models.SendOptions;

List<EventData> events = Arrays.asList(
    new EventData("Melbourne"), 
    new EventData("London"),
    new EventData("New York"));

// Events with same partition key go to same partition
SendOptions sendOptions = new SendOptions().setPartitionKey("cities");
producer.send(events, sendOptions);

EventHubConsumerClient

Synchronous Consumer

import com.azure.core.util.IterableStream;
import com.azure.messaging.eventhubs.*;
import com.azure.messaging.eventhubs.models.EventPosition;
import com.azure.messaging.eventhubs.models.PartitionEvent;
import com.azure.identity.DefaultAzureCredentialBuilder;

import java.time.Duration;
import java.time.Instant;

EventHubConsumerClient consumer = new EventHubClientBuilder()
    .credential("<<fully-qualified-namespace>>", "<<event-hub-name>>",
        new DefaultAzureCredentialBuilder().build())
    .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
    .buildConsumerClient();

// Start from 12 hours ago
Instant twelveHoursAgo = Instant.now().minus(Duration.ofHours(12));
EventPosition startingPosition = EventPosition.fromEnqueuedTime(twelveHoursAgo);
String partitionId = "0";

// Receive up to 100 events or wait 30 seconds
IterableStream<PartitionEvent> events = consumer.receiveFromPartition(
    partitionId, 100, startingPosition, Duration.ofSeconds(30));

Long lastSequenceNumber = -1L;
for (PartitionEvent partitionEvent : events) {
    System.out.print("Event received: " + partitionEvent.getData().getSequenceNumber());
    lastSequenceNumber = partitionEvent.getData().getSequenceNumber();
}

// Continue from last processed event
if (lastSequenceNumber != -1L) {
    EventPosition nextPosition = EventPosition.fromSequenceNumber(lastSequenceNumber, false);
    IterableStream<PartitionEvent> nextEvents = consumer.receiveFromPartition(
        partitionId, 100, nextPosition, Duration.ofSeconds(30));
}

consumer.close();

Asynchronous Consumer (Reactive)

import com.azure.messaging.eventhubs.*;
import com.azure.messaging.eventhubs.models.PartitionContext;
import reactor.core.Disposable;

EventHubConsumerAsyncClient consumer = new EventHubClientBuilder()
    .credential("<<fully-qualified-namespace>>", "<<event-hub-name>>",
        new DefaultAzureCredentialBuilder().build())
    .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
    .buildAsyncConsumerClient();

String partitionId = "0";
EventPosition startingPosition = EventPosition.latest();

// Non-blocking - returns immediately
Disposable subscription = consumer.receiveFromPartition(partitionId, startingPosition)
    .subscribe(partitionEvent -> {
        PartitionContext partitionContext = partitionEvent.getPartitionContext();
        EventData event = partitionEvent.getData();

        System.out.printf("Received event from partition '%s'%n", partitionContext.getPartitionId());
        System.out.printf("Contents: '%s'%n", event.getBodyAsString());
    }, error -> {
        System.err.print("An error occurred: " + error);
    }, () -> {
        System.out.print("Stream has ended.");
    });

// When done receiving
subscription.dispose();
consumer.close();

EventProcessorClient

Basic EventProcessorClient with Checkpointing

import com.azure.messaging.eventhubs.*;
import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
import com.azure.messaging.eventhubs.models.ErrorContext;
import com.azure.messaging.eventhubs.models.EventContext;
import com.azure.storage.blob.BlobContainerAsyncClient;
import com.azure.storage.blob.BlobContainerClientBuilder;
import com.azure.identity.DefaultAzureCredentialBuilder;

import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public class EventProcessorClientSample {
    public static void main(String[] args) throws Exception {
        
        // Event handler - processes each event and checkpoints
        Consumer<EventContext> processEvent = eventContext -> {
            System.out.printf("Processing event: partition=%s, sequence=%d%n",
                eventContext.getPartitionContext().getPartitionId(),
                eventContext.getEventData().getSequenceNumber());

            // Checkpoint after processing each event
            eventContext.updateCheckpoint();
        };

        // Error handler - logs errors, processor keeps running
        Consumer<ErrorContext> processError = errorContext -> {
            System.err.printf("Error while processing partition %s: %s%n", 
                errorContext.getPartitionContext().getPartitionId(),
                errorContext.getThrowable().getMessage());
        };

        // Create blob container client for checkpoint store
        BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
            .credential(new DefaultAzureCredentialBuilder().build())
            .endpoint("<storage-account-url>")
            .containerName("checkpoints")
            .buildAsyncClient();

        EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
            .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
            .credential("<<fully-qualified-namespace>>", "<<event-hub-name>>",
                new DefaultAzureCredentialBuilder().build())
            .processEvent(processEvent)
            .processError(processError)
            .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
            .buildEventProcessorClient();
        
        System.out.println("Starting event processor");
        eventProcessorClient.start();

        // Processor runs in background - do other work
        Thread.sleep(TimeUnit.MINUTES.toMillis(1));

        System.out.println("Stopping event processor");
        eventProcessorClient.stop();
    }
}

Checkpointing Patterns

Batch Processing with Periodic Checkpointing

import com.azure.messaging.eventhubs.*;
import com.azure.messaging.eventhubs.models.EventBatchContext;

import java.time.Duration;
import java.util.function.Consumer;

// Process 50 events in a batch OR wait up to 30 seconds
Consumer<EventBatchContext> processBatch = batchContext -> {
    if (batchContext.getEvents().isEmpty()) {
        return;
    }

    for (EventData event : batchContext.getEvents()) {
        System.out.printf("Processing event: partition=%s, sequence=%d%n",
            batchContext.getPartitionContext().getPartitionId(),
            event.getSequenceNumber());
    }

    // Checkpoint after processing entire batch
    batchContext.updateCheckpoint();
};

EventProcessorClient processor = new EventProcessorClientBuilder()
    .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
    .credential("<<fully-qualified-namespace>>", "<<event-hub-name>>",
        new DefaultAzureCredentialBuilder().build())
    .processEventBatch(processBatch, 50, Duration.ofSeconds(30))
    .processError(processError)
    .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
    .buildEventProcessorClient();

processor.start();

Checkpoint After N Events

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

// Track events per partition
Map<String, AtomicInteger> partitionCounters = new ConcurrentHashMap<>();
int checkpointAfterN = 100;

Consumer<EventContext> processEvent = eventContext -> {
    String partitionId = eventContext.getPartitionContext().getPartitionId();
    
    // Process event
    System.out.println("Event: " + eventContext.getEventData().getBodyAsString());
    
    // Increment counter for this partition
    AtomicInteger counter = partitionCounters.computeIfAbsent(
        partitionId, k -> new AtomicInteger(0));
    
    if (counter.incrementAndGet() >= checkpointAfterN) {
        eventContext.updateCheckpoint();
        counter.set(0);
        System.out.printf("Checkpointed partition %s%n", partitionId);
    }
};

Partition Handling

Get Partition Information

EventHubProducerClient producer = new EventHubClientBuilder()
    .credential("<<namespace>>", "<<event-hub>>",
        new DefaultAzureCredentialBuilder().build())
    .buildProducerClient();

// Get Event Hub properties
EventHubProperties eventHubProperties = producer.getEventHubProperties();
System.out.println("Event Hub: " + eventHubProperties.getName());
System.out.println("Partitions: " + eventHubProperties.getPartitionIds());

// Get specific partition properties
for (String partitionId : eventHubProperties.getPartitionIds()) {
    PartitionProperties partitionProperties = producer.getPartitionProperties(partitionId);
    System.out.printf("Partition %s: begin=%d, end=%d%n",
        partitionId,
        partitionProperties.getBeginningSequenceNumber(),
        partitionProperties.getLastEnqueuedSequenceNumber());
}

Event Position Options

import com.azure.messaging.eventhubs.models.EventPosition;
import java.time.Instant;

// From beginning of partition
EventPosition fromStart = EventPosition.earliest();

// From end (new events only)
EventPosition fromEnd = EventPosition.latest();

// From specific sequence number (exclusive)
EventPosition fromSequence = EventPosition.fromSequenceNumber(12345L, false);

// From specific sequence number (inclusive)
EventPosition fromSequenceInclusive = EventPosition.fromSequenceNumber(12345L, true);

// From specific time
EventPosition fromTime = EventPosition.fromEnqueuedTime(Instant.now().minusSeconds(3600));

// From specific offset
EventPosition fromOffset = EventPosition.fromOffset(1000L);

Source: SKILL.md on GitHub

Third-party checks

1 warning15d4 checks · Risk SAFE
  • Gen Agent Trust Hub15d

    This skill provides comprehensive patterns for building streaming applications with the Azure Event Hubs SDK for Java, incorporating security best practices like Azure Identity. It includes considerations regarding the processing of external event data, which is standard for its intended functionality.

  • Socket15d

    No alerts

  • Snyk15d

    Risk: LOW · No issues

  • Runlayer7mo

    3/3 files flagged

Provenance

Signed by skilld at e7ee58f. This ties the file your Agent reads to that commit on GitHub. It does not review the instructions.

Last checked against GitHub 20 hours ago.

Activeupdated 5 months ago

Capability

Other metadata
metadata
{
  "author": "Microsoft",
  "version": "1.0.0",
  "package": "com.azure:azure-messaging-eventhubs"
}

README badge

README badge for microsoft/skills/azure-eventhub-java