Azure Eventhub Ts

1---
2name: azure-eventhub-ts
3description: Build event streaming applications using Azure Event Hubs SDK for JavaScript (@azure/event-hubs). Use when implementing high-throughput event ingestion, real-time analytics, IoT telemetry, or event-driven architectures with partitioned consumers.
4package: "@azure/event-hubs"
5---
6 
7# Azure Event Hubs SDK for TypeScript
8 
9High-throughput event streaming and real-time data ingestion.
10 
11## Installation
12 
13```bash
14npm install @azure/event-hubs @azure/identity
15```
16 
17For checkpointing with consumer groups:
18```bash
19npm install @azure/eventhubs-checkpointstore-blob @azure/storage-blob
20```
21 
22## Environment Variables
23 
24```bash
25EVENTHUB_NAMESPACE=<namespace>.servicebus.windows.net
26EVENTHUB_NAME=my-eventhub
27STORAGE_ACCOUNT_NAME=<storage-account>
28STORAGE_CONTAINER_NAME=checkpoints
29```
30 
31## Authentication
32 
33```typescript
34import { EventHubProducerClient, EventHubConsumerClient } from "@azure/event-hubs";
35import { DefaultAzureCredential } from "@azure/identity";
36 
37const fullyQualifiedNamespace = process.env.EVENTHUB_NAMESPACE!;
38const eventHubName = process.env.EVENTHUB_NAME!;
39const credential = new DefaultAzureCredential();
40 
41// Producer
42const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);
43 
44// Consumer
45const consumer = new EventHubConsumerClient(
46  "$Default", // Consumer group
47  fullyQualifiedNamespace,
48  eventHubName,
49  credential
50);
51```
52 
53## Core Workflow
54 
55### Send Events
56 
57```typescript
58const producer = new EventHubProducerClient(namespace, eventHubName, credential);
59 
60// Create batch and add events
61const batch = await producer.createBatch();
62batch.tryAdd({ body: { temperature: 72.5, deviceId: "sensor-1" } });
63batch.tryAdd({ body: { temperature: 68.2, deviceId: "sensor-2" } });
64 
65await producer.sendBatch(batch);
66await producer.close();
67```
68 
69### Send to Specific Partition
70 
71```typescript
72// By partition ID
73const batch = await producer.createBatch({ partitionId: "0" });
74 
75// By partition key (consistent hashing)
76const batch = await producer.createBatch({ partitionKey: "device-123" });
77```
78 
79### Receive Events (Simple)
80 
81```typescript
82const consumer = new EventHubConsumerClient("$Default", namespace, eventHubName, credential);
83 
84const subscription = consumer.subscribe({
85  processEvents: async (events, context) => {
86    for (const event of events) {
87      console.log(`Partition: ${context.partitionId}, Body: ${JSON.stringify(event.body)}`);
88    }
89  },
90  processError: async (err, context) => {
91    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
92  },
93});
94 
95// Stop after some time
96setTimeout(async () => {
97  await subscription.close();
98  await consumer.close();
99}, 60000);
100```
101 
102### Receive with Checkpointing (Production)
103 
104```typescript
105import { EventHubConsumerClient } from "@azure/event-hubs";
106import { ContainerClient } from "@azure/storage-blob";
107import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";
108 
109const containerClient = new ContainerClient(
110  `https://${storageAccount}.blob.core.windows.net/${containerName}`,
111  credential
112);
113 
114const checkpointStore = new BlobCheckpointStore(containerClient);
115 
116const consumer = new EventHubConsumerClient(
117  "$Default",
118  namespace,
119  eventHubName,
120  credential,
121  checkpointStore
122);
123 
124const subscription = consumer.subscribe({
125  processEvents: async (events, context) => {
126    for (const event of events) {
127      console.log(`Processing: ${JSON.stringify(event.body)}`);
128    }
129    // Checkpoint after processing batch
130    if (events.length > 0) {
131      await context.updateCheckpoint(events[events.length - 1]);
132    }
133  },
134  processError: async (err, context) => {
135    console.error(`Error: ${err.message}`);
136  },
137});
138```
139 
140### Receive from Specific Position
141 
142```typescript
143const subscription = consumer.subscribe({
144  processEvents: async (events, context) => { /* ... */ },
145  processError: async (err, context) => { /* ... */ },
146}, {
147  startPosition: {
148    // Start from beginning
149    "0": { offset: "@earliest" },
150    // Start from end (new events only)
151    "1": { offset: "@latest" },
152    // Start from specific offset
153    "2": { offset: "12345" },
154    // Start from specific time
155    "3": { enqueuedOn: new Date("2024-01-01") },
156  },
157});
158```
159 
160## Event Hub Properties
161 
162```typescript
163// Get hub info
164const hubProperties = await producer.getEventHubProperties();
165console.log(`Partitions: ${hubProperties.partitionIds}`);
166 
167// Get partition info
168const partitionProperties = await producer.getPartitionProperties("0");
169console.log(`Last sequence: ${partitionProperties.lastEnqueuedSequenceNumber}`);
170```
171 
172## Batch Processing Options
173 
174```typescript
175const subscription = consumer.subscribe(
176  {
177    processEvents: async (events, context) => { /* ... */ },
178    processError: async (err, context) => { /* ... */ },
179  },
180  {
181    maxBatchSize: 100,           // Max events per batch
182    maxWaitTimeInSeconds: 30,    // Max wait for batch
183  }
184);
185```
186 
187## Key Types
188 
189```typescript
190import {
191  EventHubProducerClient,
192  EventHubConsumerClient,
193  EventData,
194  ReceivedEventData,
195  PartitionContext,
196  Subscription,
197  SubscriptionEventHandlers,
198  CreateBatchOptions,
199  EventPosition,
200} from "@azure/event-hubs";
201 
202import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";
203```
204 
205## Event Properties
206 
207```typescript
208// Send with properties
209const batch = await producer.createBatch();
210batch.tryAdd({
211  body: { data: "payload" },
212  properties: {
213    eventType: "telemetry",
214    deviceId: "sensor-1",
215  },
216  contentType: "application/json",
217  correlationId: "request-123",
218});
219 
220// Access in receiver
221consumer.subscribe({
222  processEvents: async (events, context) => {
223    for (const event of events) {
224      console.log(`Type: ${event.properties?.eventType}`);
225      console.log(`Sequence: ${event.sequenceNumber}`);
226      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
227      console.log(`Offset: ${event.offset}`);
228    }
229  },
230});
231```
232 
233## Error Handling
234 
235```typescript
236consumer.subscribe({
237  processEvents: async (events, context) => {
238    try {
239      for (const event of events) {
240        await processEvent(event);
241      }
242      await context.updateCheckpoint(events[events.length - 1]);
243    } catch (error) {
244      // Don't checkpoint on error - events will be reprocessed
245      console.error("Processing failed:", error);
246    }
247  },
248  processError: async (err, context) => {
249    if (err.name === "MessagingError") {
250      // Transient error - SDK will retry
251      console.warn("Transient error:", err.message);
252    } else {
253      // Fatal error
254      console.error("Fatal error:", err);
255    }
256  },
257});
258```
259 
260## Best Practices
261 
2621. **Use checkpointing** - Always checkpoint in production for exactly-once processing
2632. **Batch sends** - Use `createBatch()` for efficient sending
2643. **Partition keys** - Use partition keys to ensure ordering for related events
2654. **Consumer groups** - Use separate consumer groups for different processing pipelines
2665. **Handle errors gracefully** - Don't checkpoint on processing failures
2676. **Close clients** - Always close producer/consumer when done
2687. **Monitor lag** - Track `lastEnqueuedSequenceNumber` vs processed sequence
269