AutoGineer
Generated banner for Azure Eventhub TS
Generated logo for Azure Eventhub TS

Azure Eventhub TS

MicrosoftOfficialCreated by Microsoft

  • Skill
  • Tool integration
  • Cloud and DevOps
  • MIT
  • No issues found
Download ZIP

From the author

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.
— the description by Microsoft

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

This skill provides the necessary patterns and interfaces to interact with Azure Event Hubs using the @azure/event-hubs SDK. It covers the core functionality for producing and consuming event streams, including batch processing, partition management, and checkpointing strategies using Azure Blob Storage.

What it does

  • Establishes connections to Azure Event Hubs using identity-based authentication.
  • Supports sending events individually or in batches to specific partitions.
  • Enables consumption of event streams with configurable batch sizes and wait times.
  • Provides mechanisms for persistent checkpointing to track event processing progress.
  • Offers event handlers for initialization, error management, and event processing.
  • Allows retrieval of metadata regarding Event Hub partitions and sequence numbers.

When to use it

  • When building event-driven architectures in TypeScript.
  • When implementing high-throughput data ingestion pipelines.
  • When you need to manage event consumption across multiple partitions.
  • When integrating real-time telemetry or streaming data into your application.

Advantages

  • Native support for TypeScript and Azure identity providers.
  • Flexible subscription patterns for handling event streams.
  • Built-in support for checkpointing to manage state across restarts.
  • Granular control over partition selection and event positioning.

Usage examples

“How do I send a batch of events to an Azure Event Hub?”
The agent provides a code snippet using EventHubProducerClient to create a batch, add events with body data, and send the batch to the hub.
“Show me how to consume events with checkpointing.”
The agent provides an example using EventHubConsumerClient configured with a BlobCheckpointStore to process events and update the checkpoint after each batch.

Install

  1. Download the skill as a ZIP:

    https://mp-staging.autogineer.com/agent-skills/microsoft/azure-eventhub-ts/download

  2. Unzip it into the cross-client skills folder of your project:

    mkdir -p .agents/skills && unzip azure-eventhub-ts.zip -d .agents/skills

Documentation by Microsoft

Azure Event Hubs SDK for TypeScript

High-throughput event streaming and real-time data ingestion.

Installation

npm install @azure/event-hubs @azure/identity

For checkpointing with consumer groups:

npm install @azure/eventhubs-checkpointstore-blob @azure/storage-blob

Environment Variables

EVENTHUB_NAMESPACE=<namespace>.servicebus.windows.net
EVENTHUB_NAME=my-eventhub
STORAGE_ACCOUNT_NAME=<storage-account>
STORAGE_CONTAINER_NAME=checkpoints
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication

import { EventHubProducerClient, EventHubConsumerClient } from "@azure/event-hubs";
import { DefaultAzureCredential, ManagedIdentityCredential } from "@azure/identity";

const fullyQualifiedNamespace = process.env.EVENTHUB_NAMESPACE!;
const eventHubName = process.env.EVENTHUB_NAME!;
// Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
const credential = new DefaultAzureCredential({requiredEnvVars: ["AZURE_TOKEN_CREDENTIALS"]});
// Or use a specific credential directly in production:
// See https://learn.microsoft.com/javascript/api/overview/azure/identity-readme?view=azure-node-latest#credential-classes
// const credential = new ManagedIdentityCredential();

// Producer
const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);

// Consumer
const consumer = new EventHubConsumerClient(
  "$Default", // Consumer group
  fullyQualifiedNamespace,
  eventHubName,
  credential
);

Core Workflow

Send Events
const producer = new EventHubProducerClient(namespace, eventHubName, credential);

// Create batch and add events
const batch = await producer.createBatch();
batch.tryAdd({ body: { temperature: 72.5, deviceId: "sensor-1" } });
batch.tryAdd({ body: { temperature: 68.2, deviceId: "sensor-2" } });

await producer.sendBatch(batch);
await producer.close();
Send to Specific Partition
// By partition ID
const batch = await producer.createBatch({ partitionId: "0" });

// By partition key (consistent hashing)
const batch = await producer.createBatch({ partitionKey: "device-123" });
Receive Events (Simple)
const consumer = new EventHubConsumerClient("$Default", namespace, eventHubName, credential);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Partition: ${context.partitionId}, Body: ${JSON.stringify(event.body)}`);
    }
  },
  processError: async (err, context) => {
    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
  },
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await consumer.close();
}, 60000);
Receive with Checkpointing (Production)
import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const containerClient = new ContainerClient(
  `https://${storageAccount}.blob.core.windows.net/${containerName}`,
  credential
);

const checkpointStore = new BlobCheckpointStore(containerClient);

const consumer = new EventHubConsumerClient(
  "$Default",
  namespace,
  eventHubName,
  credential,
  checkpointStore
);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Processing: ${JSON.stringify(event.body)}`);
    }
    // Checkpoint after processing batch
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  },
});
Receive from Specific Position
const subscription = consumer.subscribe({
  processEvents: async (events, context) => { /* ... */ },
  processError: async (err, context) => { /* ... */ },
}, {
  startPosition: {
    // Start from beginning
    "0": { offset: "@earliest" },
    // Start from end (new events only)
    "1": { offset: "@latest" },
    // Start from specific offset
    "2": { offset: "12345" },
    // Start from specific time
    "3": { enqueuedOn: new Date("2024-01-01") },
  },
});

Event Hub Properties

// Get hub info
const hubProperties = await producer.getEventHubProperties();
console.log(`Partitions: ${hubProperties.partitionIds}`);

// Get partition info
const partitionProperties = await producer.getPartitionProperties("0");
console.log(`Last sequence: ${partitionProperties.lastEnqueuedSequenceNumber}`);

Batch Processing Options

const subscription = consumer.subscribe(
  {
    processEvents: async (events, context) => { /* ... */ },
    processError: async (err, context) => { /* ... */ },
  },
  {
    maxBatchSize: 100,           // Max events per batch
    maxWaitTimeInSeconds: 30,    // Max wait for batch
  }
);

Key Types

import {
  EventHubProducerClient,
  EventHubConsumerClient,
  EventData,
  ReceivedEventData,
  PartitionContext,
  Subscription,
  SubscriptionEventHandlers,
  CreateBatchOptions,
  EventPosition,
} from "@azure/event-hubs";

import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

Event Properties

// Send with properties
const batch = await producer.createBatch();
batch.tryAdd({
  body: { data: "payload" },
  properties: {
    eventType: "telemetry",
    deviceId: "sensor-1",
  },
  contentType: "application/json",
  correlationId: "request-123",
});

// Access in receiver
consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Type: ${event.properties?.eventType}`);
      console.log(`Sequence: ${event.sequenceNumber}`);
      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
      console.log(`Offset: ${event.offset}`);
    }
  },
});

Error Handling

consumer.subscribe({
  processEvents: async (events, context) => {
    try {
      for (const event of events) {
        await processEvent(event);
      }
      await context.updateCheckpoint(events[events.length - 1]);
    } catch (error) {
      // Don't checkpoint on error - events will be reprocessed
      console.error("Processing failed:", error);
    }
  },
  processError: async (err, context) => {
    if (err.name === "MessagingError") {
      // Transient error - SDK will retry
      console.warn("Transient error:", err.message);
    } else {
      // Fatal error
      console.error("Fatal error:", err);
    }
  },
});

Best Practices

  1. Use checkpointing - Always checkpoint in production for exactly-once processing

  2. Batch sends - Use createBatch() for efficient sending

  3. Partition keys - Use partition keys to ensure ordering for related events

  4. Consumer groups - Use separate consumer groups for different processing pipelines

  5. Handle errors gracefully - Don't checkpoint on processing failures

  6. Close clients - Always close producer/consumer when done

  7. Monitor lag - Track lastEnqueuedSequenceNumber vs processed sequence

Files

references/checkpointing.md

Checkpointing Reference

Persistent checkpointing with BlobCheckpointStore for Azure Event Hubs consumer applications.

Overview

Checkpointing tracks the last successfully processed event position per partition. This enables:

  • Resumption — Continue from where you left off after restart

  • Load balancing — Distribute partitions across multiple consumers

  • Exactly-once processing — When combined with idempotent downstream operations

Installation

npm install @azure/event-hubs @azure/eventhubs-checkpointstore-blob @azure/storage-blob @azure/identity

Key Interfaces

import { CheckpointStore, Checkpoint, PartitionOwnership } from "@azure/event-hubs";

// CheckpointStore interface (implemented by BlobCheckpointStore)
interface CheckpointStore {
  listCheckpoints(
    fullyQualifiedNamespace: string,
    eventHubName: string,
    consumerGroup: string
  ): Promise<Checkpoint[]>;

  updateCheckpoint(checkpoint: Checkpoint): Promise<void>;

  listOwnership(
    fullyQualifiedNamespace: string,
    eventHubName: string,
    consumerGroup: string
  ): Promise<PartitionOwnership[]>;

  claimOwnership(
    partitionOwnership: PartitionOwnership[]
  ): Promise<PartitionOwnership[]>;
}

// Checkpoint structure
interface Checkpoint {
  fullyQualifiedNamespace: string;
  eventHubName: string;
  consumerGroup: string;
  partitionId: string;
  sequenceNumber: number;
  offset: string;
}

// Partition ownership (for load balancing)
interface PartitionOwnership {
  fullyQualifiedNamespace: string;
  eventHubName: string;
  consumerGroup: string;
  partitionId: string;
  ownerId: string;
  lastModifiedTimeInMs?: number;
  etag?: string;
}

Basic Checkpointing Setup

import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";
import { DefaultAzureCredential } from "@azure/identity";

const credential = new DefaultAzureCredential();

// Storage for checkpoints
const storageAccountName = process.env.STORAGE_ACCOUNT_NAME!;
const containerName = "eventhub-checkpoints";

const containerClient = new ContainerClient(
  `https://${storageAccountName}.blob.core.windows.net/${containerName}`,
  credential
);

// Ensure container exists
await containerClient.createIfNotExists();

// Create checkpoint store
const checkpointStore = new BlobCheckpointStore(containerClient);

// Create consumer with checkpoint store
const consumerClient = new EventHubConsumerClient(
  "$Default",
  "my-namespace.servicebus.windows.net",
  "my-event-hub",
  credential,
  checkpointStore  // Pass checkpoint store here
);

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
    }
    
    // Checkpoint after successful processing
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  }
});

Connection String Authentication

import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

// Storage connection string
const storageConnectionString = process.env.STORAGE_CONNECTION_STRING!;
const containerName = "checkpoints";

const containerClient = new ContainerClient(
  storageConnectionString,
  containerName
);

await containerClient.createIfNotExists();

const checkpointStore = new BlobCheckpointStore(containerClient);

// Event Hub connection string
const eventHubConnectionString = process.env.EVENTHUB_CONNECTION_STRING!;
const eventHubName = process.env.EVENTHUB_NAME!;

const consumerClient = new EventHubConsumerClient(
  "$Default",
  eventHubConnectionString,
  eventHubName,
  checkpointStore
);

Checkpointing Strategies

Checkpoint Every Batch (Default)
const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
    }
    
    // Checkpoint after every batch
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
Checkpoint Every N Events
let processedCount = 0;
const checkpointInterval = 100;

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
      processedCount++;
      
      // Checkpoint every N events
      if (processedCount % checkpointInterval === 0) {
        await context.updateCheckpoint(event);
        console.log(`Checkpointed at ${event.sequenceNumber}`);
      }
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
Checkpoint on Time Interval
let lastCheckpointTime = Date.now();
let lastEvent: ReceivedEventData | null = null;
const checkpointIntervalMs = 30000; // 30 seconds

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
      lastEvent = event;
    }
    
    // Checkpoint based on time
    const now = Date.now();
    if (lastEvent && (now - lastCheckpointTime) >= checkpointIntervalMs) {
      await context.updateCheckpoint(lastEvent);
      lastCheckpointTime = now;
      console.log(`Time-based checkpoint at ${lastEvent.sequenceNumber}`);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
Checkpoint Only on Success
const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    const results = await Promise.allSettled(
      events.map(event => processEvent(event))
    );
    
    // Find last successfully processed event
    let lastSuccessIndex = -1;
    for (let i = results.length - 1; i >= 0; i--) {
      if (results[i].status === "fulfilled") {
        lastSuccessIndex = i;
        break;
      }
    }
    
    // Only checkpoint up to last success
    if (lastSuccessIndex >= 0) {
      await context.updateCheckpoint(events[lastSuccessIndex]);
    }
    
    // Log failures
    const failures = results.filter(r => r.status === "rejected");
    if (failures.length > 0) {
      console.error(`${failures.length} events failed processing`);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

Load Balancing Across Consumers

Multiple consumers with the same consumer group automatically balance partitions:

// Consumer Instance 1
const consumer1 = new EventHubConsumerClient(
  "my-consumer-group",
  namespace,
  eventHubName,
  credential,
  checkpointStore
);

// Consumer Instance 2 (separate process)
const consumer2 = new EventHubConsumerClient(
  "my-consumer-group",  // Same consumer group
  namespace,
  eventHubName,
  credential,
  checkpointStore       // Same checkpoint store
);

// Both subscribe - partitions automatically distributed
consumer1.subscribe({ /* handlers */ });
consumer2.subscribe({ /* handlers */ });

Checkpoint Store Blob Structure

The BlobCheckpointStore creates blobs in this structure:

<container>/
├── <namespace>/<eventhub>/<consumergroup>/checkpoint/
│   ├── 0    (checkpoint for partition 0)
│   ├── 1    (checkpoint for partition 1)
│   └── ...
└── <namespace>/<eventhub>/<consumergroup>/ownership/
    ├── 0    (ownership for partition 0)
    ├── 1    (ownership for partition 1)
    └── ...

Inspecting Checkpoints

import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const checkpointStore = new BlobCheckpointStore(containerClient);

// List all checkpoints
const checkpoints = await checkpointStore.listCheckpoints(
  "my-namespace.servicebus.windows.net",
  "my-event-hub",
  "$Default"
);

for (const cp of checkpoints) {
  console.log(`Partition ${cp.partitionId}:`);
  console.log(`  Sequence: ${cp.sequenceNumber}`);
  console.log(`  Offset: ${cp.offset}`);
}

// List partition ownership
const ownerships = await checkpointStore.listOwnership(
  "my-namespace.servicebus.windows.net",
  "my-event-hub",
  "$Default"
);

for (const own of ownerships) {
  console.log(`Partition ${own.partitionId}: owned by ${own.ownerId}`);
}

Error Handling

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    try {
      for (const event of events) {
        await processEvent(event);
      }
      
      if (events.length > 0) {
        await context.updateCheckpoint(events[events.length - 1]);
      }
    } catch (processingError) {
      // Don't checkpoint on error - events will be reprocessed
      console.error("Processing failed:", processingError);
      // Optionally: send to dead letter, alert, etc.
    }
  },
  
  processError: async (err, context) => {
    // SDK-level errors (connection, auth, etc.)
    console.error(`SDK error on partition ${context.partitionId}:`, err.message);
    
    // The SDK handles reconnection automatically
    // Consider alerting for persistent errors
  }
});

Graceful Shutdown with Final Checkpoint

let lastProcessedEvent: Map<string, ReceivedEventData> = new Map();

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
      lastProcessedEvent.set(context.partitionId, event);
    }
    
    // Regular checkpointing
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

// Graceful shutdown
process.on("SIGTERM", async () => {
  console.log("Shutting down gracefully...");
  
  // Close subscription (stops receiving)
  await subscription.close();
  
  // Close consumer client
  await consumerClient.close();
  
  console.log("Shutdown complete");
  process.exit(0);
});

Best Practices

  1. Checkpoint after processing — Never before, to avoid data loss

  2. Balance checkpoint frequency — Too often = high storage costs; too rare = more reprocessing on restart

  3. Use same checkpoint store — All consumers in a group must share the same store

  4. Create container before use — Call createIfNotExists() on startup

  5. Handle checkpoint failures — Log and alert if updateCheckpoint fails repeatedly

  6. Use consumer groups — Separate groups for different processing pipelines

  7. Design for reprocessing — Make downstream operations idempotent

Checkpoint Frequency Trade-offs

FrequencyProsCons
Every eventMinimal reprocessingHigh storage I/O
Every batchGood balanceSome reprocessing possible
Time-basedPredictable I/OVariable reprocessing
Every N eventsControlled I/OMay miss some on crash

See Also

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs Node.js

Security

No issues found by our static scan· No issues found

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by
Microsoft
Publisher
Microsoft (Official)
Source
microsoft/skills · .github/plugins/azure-sdk-typescript/skills/azure-eventhub-ts · commit 0696600
Licence
MIT License (redistribution allowed)
  • LICENSE (added by AutoGineer from LICENSE)
  • ATTRIBUTION.md (written by AutoGineer)
Synced

AutoGineer is not affiliated with or endorsed by Microsoft.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-eventhub-ts.zip