Azure Eventhub Java

import com.azure.messaging.eventhubs.EventHubProducerClient;import com.azure.messaging.eventhubs.EventHubClientBuilder;
// With connection stringEventHubProducerClient producer = new EventHubClientBuilder()    .connectionString("<connection-string>", "<event-hub-name>")    .buildProducerClient();
// Full connection string with EntityPathEventHubProducerClient producer = new EventHubClientBuilder()    .connectionString("<connection-string-with-entity-path>")    .buildProducerClient();

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();

import com.azure.messaging.eventhubs.EventHubConsumerClient;
EventHubConsumerClient consumer = new EventHubClientBuilder()    .connectionString("<connection-string>", "<event-hub-name>")    .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)    .buildConsumerClient();

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();

import com.azure.messaging.eventhubs.EventData;
EventData eventData = new EventData("Hello, Event Hubs!");producer.send(Collections.singletonList(eventData));

import com.azure.messaging.eventhubs.EventDataBatch;import com.azure.messaging.eventhubs.models.CreateBatchOptions;
// Create batchEventDataBatch 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 eventsif (batch.getCount() > 0) {    producer.send(batch);}

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

CreateBatchOptions options = new CreateBatchOptions()    .setPartitionKey("customer-123");
EventDataBatch batch = producer.createBatch(options);batch.tryAdd(new EventData("Customer event"));producer.send(batch);

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

import com.azure.messaging.eventhubs.models.EventPosition;import com.azure.messaging.eventhubs.models.PartitionEvent;
// Receive from specific partitionIterable<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());}

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 storeBlobContainerAsyncClient blobClient = new BlobContainerClientBuilder()    .connectionString("<storage-connection-string>")    .containerName("checkpoints")    .buildAsyncClient();
// Create processorEventProcessorClient 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 processingprocessor.start();
// Keep running...Thread.sleep(Duration.ofMinutes(5).toMillis());
// Stop gracefullyprocessor.stop();

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();

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 hub infoEventHubProperties hubProps = producer.getEventHubProperties();System.out.println("Hub: " + hubProps.getName());System.out.println("Partitions: " + hubProps.getPartitionIds());
// Get partition infoPartitionProperties 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());

// Start from beginningEventPosition.earliest()
// Start from end (new events only)EventPosition.latest()
// From specific offsetEventPosition.fromOffset(12345L)
// From specific sequence numberEventPosition.fromSequenceNumber(100L)
// From specific timeEventPosition.fromEnqueuedTime(Instant.now().minus(Duration.ofHours(1)))

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());})

// Always close clientstry {    producer.send(batch);} finally {    producer.close();}
// Or use try-with-resourcestry (EventHubProducerClient producer = new EventHubClientBuilder()        .connectionString(connectionString, eventHubName)        .buildProducerClient()) {    producer.send(events);}