# SkillPatch skill: azure-eventhub-java

This skill guides agents in building real-time streaming applications using the Azure Event Hubs SDK for Java. It covers client creation (producer, consumer, sync/async), authentication via connection strings or DefaultAzureCredential, and core patterns like sending single events and batched events. It targets high-throughput data ingestion and event-driven architectures on Azure.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/azure-eventhub-java
curl -sSL https://skillpatch.dev/install_skill/azure-eventhub-java | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (2)

- `SKILL.md`
- `references/examples.md`


### `SKILL.md`

````markdown
---
name: azure-eventhub-java
description: 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.
license: MIT
metadata:
  author: Microsoft
  version: "1.0.0"
  package: com.azure:azure-messaging-eventhubs
---

# Azure Event Hubs SDK for Java

Build real-time streaming applications using the Azure Event Hubs SDK for Java.

## Installation

```xml
<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-eventhubs</artifactId>
    <version>5.19.0</version>
</dependency>

<!-- For checkpoint store (production) -->
<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
    <version>1.20.0</version>
</dependency>
```

## Client Creation

### EventHubProducerClient

```java
import com.azure.messaging.eventhubs.EventHubProducerClient;
import com.azure.messaging.eventhubs.EventHubClientBuilder;

// With connection string
EventHubProducerClient producer = new EventHubClientBuilder()
    .connectionString("<connection-string>", "<event-hub-name>")
    .buildProducerClient();

// Full connection string with EntityPath
EventHubProducerClient producer = new EventHubClientBuilder()
    .connectionString("<connection-string-with-entity-path>")
    .buildProducerClient();
```

### With DefaultAzureCredential

```java
import com.azure.core.credential.TokenCredential;
import com.azure.identity.AzureIdentityEnvVars;
import com.azure.identity.DefaultAzureCredentialBuilder;
import com.azure.identity.ManagedIdentityCredentialBuilder;

// Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
TokenCredential credential = new DefaultAzureCredentialBuilder()
    .requireEnvVars(AzureIdentityEnvVars.AZURE_TOKEN_CREDENTIALS)
    .build();
// Or use a specific credential directly in production:
// See https://learn.microsoft.com/java/api/overview/azure/identity-readme?view=azure-java-stable#credential-classes
// TokenCredential credential = new ManagedIdentityCredentialBuilder().build();

EventHubProducerClient producer = new EventHubClientBuilder()
    .fullyQualifiedNamespace("<namespace>.servicebus.windows.net")
    .eventHubName("<event-hub-name>")
    .credential(credential)
    .buildProducerClient();
```

### EventHubConsumerClient

```java
import com.azure.messaging.eventhubs.EventHubConsumerClient;

EventHubConsumerClient consumer = new EventHubClientBuilder()
    .connectionString("<connection-string>", "<event-hub-name>")
    .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
    .buildConsumerClient();
```

### Async Clients

```java
import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;

EventHubProducerAsyncClient asyncProducer = new EventHubClientBuilder()
    .connectionString("<connection-string>", "<event-hub-name>")
    .buildAsyncProducerClient();

EventHubConsumerAsyncClient asyncConsumer = new EventHubClientBuilder()
    .connectionString("<connection-string>", "<event-hub-name>")
    .consumerGroup("$Default")
    .buildAsyncConsumerClient();
```

## Core Patterns

### Send Single Event

```java
import com.azure.messaging.eventhubs.EventData;

EventData eventData = new EventData("Hello, Event Hubs!");
producer.send(Collections.singletonList(eventData));
```

### Send Event Batch

```java
import com.azure.messaging.eventhubs.EventDataBatch;
import com.azure.messaging.eventhubs.models.CreateBatchOptions;

// Create batch
EventDataBatch batch = producer.createBatch();

// Add events (returns false if batch is full)
for (int i = 0; i < 100; i++) {
    EventData event = new EventData("Event " + i);
    if (!batch.tryAdd(event)) {
        // Batch is full, send and create new batch
        producer.send(batch);
        batch = producer.createBatch();
        batch.tryAdd(event);
    }
}

// Send remaining events
if (batch.getCount() > 0) {
    producer.send(batch);
}
```

### Send to Specific Partition

```java
CreateBatchOptions options = new CreateBatchOptions()
    .setPartitionId("0");

EventDataBatch batch = producer.createBatch(options);
batch.tryAdd(new EventData("Partition 0 event"));
producer.send(batch);
```

### Send with Partition Key

```java
CreateBatchOptions options = new CreateBatchOptions()
    .setPartitionKey("customer-123");

EventDataBatch batch = producer.createBatch(options);
batch.tryAdd(new EventData("Customer event"));
producer.send(batch);
```

### Event with Properties

```java
EventData event = new EventData("Order created");
event.getProperties().put("orderId", "ORD-123");
event.getProperties().put("customerId", "CUST-456");
event.getProperties().put("priority", 1);

producer.send(Collections.singletonList(event));
```

### Receive Events (Simple)

```java
import com.azure.messaging.eventhubs.models.EventPosition;
import com.azure.messaging.eventhubs.models.PartitionEvent;

// Receive from specific partition
Iterable<PartitionEvent> events = consumer.receiveFromPartition(
    "0",                           // partitionId
    10,                            // maxEvents
    EventPosition.earliest(),      // startingPosition
    Duration.ofSeconds(30)         // timeout
);

for (PartitionEvent partitionEvent : events) {
    EventData event = partitionEvent.getData();
    System.out.println("Body: " + event.getBodyAsString());
    System.out.println("Sequence: " + event.getSequenceNumber());
    System.out.println("Offset: " + event.getOffset());
}
```

### EventProcessorClient (Production)

```java
import com.azure.messaging.eventhubs.EventProcessorClient;
import com.azure.messaging.eventhubs.EventProcessorClientBuilder;
import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
import com.azure.storage.blob.BlobContainerAsyncClient;
import com.azure.storage.blob.BlobContainerClientBuilder;

// Create checkpoint store
BlobContainerAsyncClient blobClient = new BlobContainerClientBuilder()
    .connectionString("<storage-connection-string>")
    .containerName("checkpoints")
    .buildAsyncClient();

// Create processor
EventProcessorClient processor = new EventProcessorClientBuilder()
    .connectionString("<eventhub-connection-string>", "<event-hub-name>")
    .consumerGroup("$Default")
    .checkpointStore(new BlobCheckpointStore(blobClient))
    .processEvent(eventContext -> {
        EventData event = eventContext.getEventData();
        System.out.println("Processing: " + event.getBodyAsString());
        
        // Checkpoint after processing
        eventContext.updateCheckpoint();
    })
    .processError(errorContext -> {
        System.err.println("Error: " + errorContext.getThrowable().getMessage());
        System.err.println("Partition: " + errorContext.getPartitionContext().getPartitionId());
    })
    .buildEventProcessorClient();

// Start processing
processor.start();

// Keep running...
Thread.sleep(Duration.ofMinutes(5).toMillis());

// Stop gracefully
processor.stop();
```

### Batch Processing

```java
EventProcessorClient processor = new EventProcessorClientBuilder()
    .connectionString("<connection-string>", "<event-hub-name>")
    .consumerGroup("$Default")
    .checkpointStore(new BlobCheckpointStore(blobClient))
    .processEventBatch(eventBatchContext -> {
        List<EventData> events = eventBatchContext.getEvents();
        System.out.printf("Received %d events%n", events.size());
        
        for (EventData event : events) {
            // Process each event
            System.out.println(event.getBodyAsString());
        }
        
        // Checkpoint after batch
        eventBatchContext.updateCheckpoint();
    }, 50) // maxBatchSize
    .processError(errorContext -> {
        System.err.println("Error: " + errorContext.getThrowable());
    })
    .buildEventProcessorClient();
```

### Async Receiving

```java
asyncConsumer.receiveFromPartition("0", EventPosition.latest())
    .subscribe(
        partitionEvent -> {
            EventData event = partitionEvent.getData();
            System.out.println("Received: " + event.getBodyAsString());
        },
        error -> System.err.println("Error: " + error),
        () -> System.out.println("Complete")
    );
```

### Get Event Hub Properties

```java
// Get hub info
EventHubProperties hubProps = producer.getEventHubProperties();
System.out.println("Hub: " + hubProps.getName());
System.out.println("Partitions: " + hubProps.getPartitionIds());

// Get partition info
PartitionProperties partitionProps = producer.getPartitionProperties("0");
System.out.println("Begin sequence: " + partitionProps.getBeginningSequenceNumber());
System.out.println("Last sequence: " + partitionProps.getLastEnqueuedSequenceNumber());
System.out.println("Last offset: " + partitionProps.getLastEnqueuedOffset());
```

## Event Positions

```java
// Start from beginning
EventPosition.earliest()

// Start from end (new events only)
EventPosition.latest()

// From specific offset
EventPosition.fromOffset(12345L)

// From specific sequence number
EventPosition.fromSequenceNumber(100L)

// From specific time
EventPosition.fromEnqueuedTime(Instant.now().minus(Duration.ofHours(1)))
```

## Error Handling

```java
import com.azure.messaging.eventhubs.models.ErrorContext;

.processError(errorContext -> {
    Throwable error = errorContext.getThrowable();
    String partitionId = errorContext.getPartitionContext().getPartitionId();
    
    if (error instanceof AmqpException) {
        AmqpException amqpError = (AmqpException) error;
        if (amqpError.isTransient()) {
            System.out.println("Transient error, will retry");
        }
    }
    
    System.err.printf("Error on partition %s: %s%n", partitionId, error.getMessage());
})
```

## Resource Cleanup

```java
// Always close clients
try {
    producer.send(batch);
} finally {
    producer.close();
}

// Or use try-with-resources
try (EventHubProducerClient producer = new EventHubClientBuilder()
        .connectionString(connectionString, eventHubName)
        .buildProducerClient()) {
    producer.send(events);
}
```

## Environment Variables

```bash
EVENT_HUBS_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...  # Alternative to Entra ID auth
EVENT_HUBS_NAME=<event-hub-name>  # Required for event hub name
STORAGE_CONNECTION_STRING=<for-checkpointing>  # Alternative to Entra ID auth for checkpointing
AZURE_TOKEN_CREDENTIALS=prod  # Required only if DefaultAzureCredential is used in production
```

## Best Practices

1. **Use EventProcessorClient**: For production, provides load balancing and checkpointing
2. **Batch Events**: Use `EventDataBatch` for efficient sending
3. **Partition Keys**: Use for ordering guarantees within a partition
4. **Checkpointing**: Checkpoint after processing to avoid reprocessing
5. **Error Handling**: Handle transient errors with retries
6. **Close Clients**: Always close producer/consumer when done

## Trigger Phrases

- "Event Hubs Java"
- "event streaming Azure"
- "real-time data ingestion"
- "EventProcessorClient"
- "event hub producer consumer"
- "partition processing"

````


### `references/examples.md`

````markdown
# Azure Event Hubs Java SDK - Examples

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

## Table of Contents

- [Maven Dependency](#maven-dependency)
- [EventHubProducerClient](#eventhubproducerclient)
- [EventHubConsumerClient](#eventhubconsumerclient)
- [EventProcessorClient](#eventprocessorclient)
- [Checkpointing Patterns](#checkpointing-patterns)
- [Partition Handling](#partition-handling)

---

## Maven Dependency

```xml
<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

```java
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

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

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

### Send to Specific Partition

```java
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)

```java
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

```java
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)

```java
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

```java
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

```java
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

```java
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

```java
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

```java
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);
```

````