Azure Eventhub Java

1---
2name: azure-eventhub-java
3description: 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.
4package: com.azure:azure-messaging-eventhubs
5---
6 
7# Azure Event Hubs SDK for Java
8 
9Build real-time streaming applications using the Azure Event Hubs SDK for Java.
10 
11## Installation
12 
13```xml
14<dependency>
15    <groupId>com.azure</groupId>
16    <artifactId>azure-messaging-eventhubs</artifactId>
17    <version>5.19.0</version>
18</dependency>
19 
20<!-- For checkpoint store (production) -->
21<dependency>
22    <groupId>com.azure</groupId>
23    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
24    <version>1.20.0</version>
25</dependency>
26```
27 
28## Client Creation
29 
30### EventHubProducerClient
31 
32```java
33import com.azure.messaging.eventhubs.EventHubProducerClient;
34import com.azure.messaging.eventhubs.EventHubClientBuilder;
35 
36// With connection string
37EventHubProducerClient producer = new EventHubClientBuilder()
38    .connectionString("<connection-string>", "<event-hub-name>")
39    .buildProducerClient();
40 
41// Full connection string with EntityPath
42EventHubProducerClient producer = new EventHubClientBuilder()
43    .connectionString("<connection-string-with-entity-path>")
44    .buildProducerClient();
45```
46 
47### With DefaultAzureCredential
48 
49```java
50import com.azure.identity.DefaultAzureCredentialBuilder;
51 
52EventHubProducerClient producer = new EventHubClientBuilder()
53    .fullyQualifiedNamespace("<namespace>.servicebus.windows.net")
54    .eventHubName("<event-hub-name>")
55    .credential(new DefaultAzureCredentialBuilder().build())
56    .buildProducerClient();
57```
58 
59### EventHubConsumerClient
60 
61```java
62import com.azure.messaging.eventhubs.EventHubConsumerClient;
63 
64EventHubConsumerClient consumer = new EventHubClientBuilder()
65    .connectionString("<connection-string>", "<event-hub-name>")
66    .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
67    .buildConsumerClient();
68```
69 
70### Async Clients
71 
72```java
73import com.azure.messaging.eventhubs.EventHubProducerAsyncClient;
74import com.azure.messaging.eventhubs.EventHubConsumerAsyncClient;
75 
76EventHubProducerAsyncClient asyncProducer = new EventHubClientBuilder()
77    .connectionString("<connection-string>", "<event-hub-name>")
78    .buildAsyncProducerClient();
79 
80EventHubConsumerAsyncClient asyncConsumer = new EventHubClientBuilder()
81    .connectionString("<connection-string>", "<event-hub-name>")
82    .consumerGroup("$Default")
83    .buildAsyncConsumerClient();
84```
85 
86## Core Patterns
87 
88### Send Single Event
89 
90```java
91import com.azure.messaging.eventhubs.EventData;
92 
93EventData eventData = new EventData("Hello, Event Hubs!");
94producer.send(Collections.singletonList(eventData));
95```
96 
97### Send Event Batch
98 
99```java
100import com.azure.messaging.eventhubs.EventDataBatch;
101import com.azure.messaging.eventhubs.models.CreateBatchOptions;
102 
103// Create batch
104EventDataBatch batch = producer.createBatch();
105 
106// Add events (returns false if batch is full)
107for (int i = 0; i < 100; i++) {
108    EventData event = new EventData("Event " + i);
109    if (!batch.tryAdd(event)) {
110        // Batch is full, send and create new batch
111        producer.send(batch);
112        batch = producer.createBatch();
113        batch.tryAdd(event);
114    }
115}
116 
117// Send remaining events
118if (batch.getCount() > 0) {
119    producer.send(batch);
120}
121```
122 
123### Send to Specific Partition
124 
125```java
126CreateBatchOptions options = new CreateBatchOptions()
127    .setPartitionId("0");
128 
129EventDataBatch batch = producer.createBatch(options);
130batch.tryAdd(new EventData("Partition 0 event"));
131producer.send(batch);
132```
133 
134### Send with Partition Key
135 
136```java
137CreateBatchOptions options = new CreateBatchOptions()
138    .setPartitionKey("customer-123");
139 
140EventDataBatch batch = producer.createBatch(options);
141batch.tryAdd(new EventData("Customer event"));
142producer.send(batch);
143```
144 
145### Event with Properties
146 
147```java
148EventData event = new EventData("Order created");
149event.getProperties().put("orderId", "ORD-123");
150event.getProperties().put("customerId", "CUST-456");
151event.getProperties().put("priority", 1);
152 
153producer.send(Collections.singletonList(event));
154```
155 
156### Receive Events (Simple)
157 
158```java
159import com.azure.messaging.eventhubs.models.EventPosition;
160import com.azure.messaging.eventhubs.models.PartitionEvent;
161 
162// Receive from specific partition
163Iterable<PartitionEvent> events = consumer.receiveFromPartition(
164    "0",                           // partitionId
165    10,                            // maxEvents
166    EventPosition.earliest(),      // startingPosition
167    Duration.ofSeconds(30)         // timeout
168);
169 
170for (PartitionEvent partitionEvent : events) {
171    EventData event = partitionEvent.getData();
172    System.out.println("Body: " + event.getBodyAsString());
173    System.out.println("Sequence: " + event.getSequenceNumber());
174    System.out.println("Offset: " + event.getOffset());
175}
176```
177 
178### EventProcessorClient (Production)
179 
180```java
181import com.azure.messaging.eventhubs.EventProcessorClient;
182import com.azure.messaging.eventhubs.EventProcessorClientBuilder;
183import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
184import com.azure.storage.blob.BlobContainerAsyncClient;
185import com.azure.storage.blob.BlobContainerClientBuilder;
186 
187// Create checkpoint store
188BlobContainerAsyncClient blobClient = new BlobContainerClientBuilder()
189    .connectionString("<storage-connection-string>")
190    .containerName("checkpoints")
191    .buildAsyncClient();
192 
193// Create processor
194EventProcessorClient processor = new EventProcessorClientBuilder()
195    .connectionString("<eventhub-connection-string>", "<event-hub-name>")
196    .consumerGroup("$Default")
197    .checkpointStore(new BlobCheckpointStore(blobClient))
198    .processEvent(eventContext -> {
199        EventData event = eventContext.getEventData();
200        System.out.println("Processing: " + event.getBodyAsString());
201        
202        // Checkpoint after processing
203        eventContext.updateCheckpoint();
204    })
205    .processError(errorContext -> {
206        System.err.println("Error: " + errorContext.getThrowable().getMessage());
207        System.err.println("Partition: " + errorContext.getPartitionContext().getPartitionId());
208    })
209    .buildEventProcessorClient();
210 
211// Start processing
212processor.start();
213 
214// Keep running...
215Thread.sleep(Duration.ofMinutes(5).toMillis());
216 
217// Stop gracefully
218processor.stop();
219```
220 
221### Batch Processing
222 
223```java
224EventProcessorClient processor = new EventProcessorClientBuilder()
225    .connectionString("<connection-string>", "<event-hub-name>")
226    .consumerGroup("$Default")
227    .checkpointStore(new BlobCheckpointStore(blobClient))
228    .processEventBatch(eventBatchContext -> {
229        List<EventData> events = eventBatchContext.getEvents();
230        System.out.printf("Received %d events%n", events.size());
231        
232        for (EventData event : events) {
233            // Process each event
234            System.out.println(event.getBodyAsString());
235        }
236        
237        // Checkpoint after batch
238        eventBatchContext.updateCheckpoint();
239    }, 50) // maxBatchSize
240    .processError(errorContext -> {
241        System.err.println("Error: " + errorContext.getThrowable());
242    })
243    .buildEventProcessorClient();
244```
245 
246### Async Receiving
247 
248```java
249asyncConsumer.receiveFromPartition("0", EventPosition.latest())
250    .subscribe(
251        partitionEvent -> {
252            EventData event = partitionEvent.getData();
253            System.out.println("Received: " + event.getBodyAsString());
254        },
255        error -> System.err.println("Error: " + error),
256        () -> System.out.println("Complete")
257    );
258```
259 
260### Get Event Hub Properties
261 
262```java
263// Get hub info
264EventHubProperties hubProps = producer.getEventHubProperties();
265System.out.println("Hub: " + hubProps.getName());
266System.out.println("Partitions: " + hubProps.getPartitionIds());
267 
268// Get partition info
269PartitionProperties partitionProps = producer.getPartitionProperties("0");
270System.out.println("Begin sequence: " + partitionProps.getBeginningSequenceNumber());
271System.out.println("Last sequence: " + partitionProps.getLastEnqueuedSequenceNumber());
272System.out.println("Last offset: " + partitionProps.getLastEnqueuedOffset());
273```
274 
275## Event Positions
276 
277```java
278// Start from beginning
279EventPosition.earliest()
280 
281// Start from end (new events only)
282EventPosition.latest()
283 
284// From specific offset
285EventPosition.fromOffset(12345L)
286 
287// From specific sequence number
288EventPosition.fromSequenceNumber(100L)
289 
290// From specific time
291EventPosition.fromEnqueuedTime(Instant.now().minus(Duration.ofHours(1)))
292```
293 
294## Error Handling
295 
296```java
297import com.azure.messaging.eventhubs.models.ErrorContext;
298 
299.processError(errorContext -> {
300    Throwable error = errorContext.getThrowable();
301    String partitionId = errorContext.getPartitionContext().getPartitionId();
302    
303    if (error instanceof AmqpException) {
304        AmqpException amqpError = (AmqpException) error;
305        if (amqpError.isTransient()) {
306            System.out.println("Transient error, will retry");
307        }
308    }
309    
310    System.err.printf("Error on partition %s: %s%n", partitionId, error.getMessage());
311})
312```
313 
314## Resource Cleanup
315 
316```java
317// Always close clients
318try {
319    producer.send(batch);
320} finally {
321    producer.close();
322}
323 
324// Or use try-with-resources
325try (EventHubProducerClient producer = new EventHubClientBuilder()
326        .connectionString(connectionString, eventHubName)
327        .buildProducerClient()) {
328    producer.send(events);
329}
330```
331 
332## Environment Variables
333 
334```bash
335EVENT_HUBS_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...
336EVENT_HUBS_NAME=<event-hub-name>
337STORAGE_CONNECTION_STRING=<for-checkpointing>
338```
339 
340## Best Practices
341 
3421. **Use EventProcessorClient**: For production, provides load balancing and checkpointing
3432. **Batch Events**: Use `EventDataBatch` for efficient sending
3443. **Partition Keys**: Use for ordering guarantees within a partition
3454. **Checkpointing**: Checkpoint after processing to avoid reprocessing
3465. **Error Handling**: Handle transient errors with retries
3476. **Close Clients**: Always close producer/consumer when done
348 
349## Trigger Phrases
350 
351- "Event Hubs Java"
352- "event streaming Azure"
353- "real-time data ingestion"
354- "EventProcessorClient"
355- "event hub producer consumer"
356- "partition processing"
357