Building Scalable Event Processing with Azure Event Hub

When your system needs to handle millions of events per minute, choosing the right architecture becomes critical. In this post, I’ll share how we built a high-throughput event processing system using Azure Event Hub that handles 1M+ events per 5 minutes with room to spare.

The Challenge

Our multi-tenant SaaS platform needed to:

• Process 3,333 events/second sustained (with 5,000-10,000 peak bursts)
• Scale horizontally across Kubernetes pods
• Maintain exactly-once processing guarantees
• Handle pod failures gracefully without losing events

Architecture Overview

flowchart TB
    subgraph EH["Azure Event Hub (32 partitions)"]
        P1[Partition 0-10]
        P2[Partition 11-21]
        P3[Partition 22-31]
    end
    
    subgraph Pods["Kubernetes Pods"]
        Pod1[Pod 1]
        Pod2[Pod 2]
        Pod3[Pod 3]
    end
    
    subgraph Data["Data Layer"]
        PG[(PostgreSQL)]
        MY[(MySQL POS)]
        RD[(Redis)]
    end
    
    P1 --> Pod1
    P2 --> Pod2
    P3 --> Pod3
    
    Pod1 --> PG
    Pod2 --> MY
    Pod3 --> RD
    
    style EH fill:#112240,stroke:#64ffda,stroke-width:2px
    style Pods fill:#112240,stroke:#64ffda,stroke-width:2px

Key Design Decisions

32 Partitions: Each partition handles ~1MB/sec, giving us 32,000 events/sec capacity — 10x our target load.

Consumer Groups: All pods share the same consumer group, allowing Event Hub to automatically distribute partitions.

Horizontal Pod Autoscaling: Kubernetes HPA scales from 10-20 pods based on CPU (70%) and memory (80%) thresholds.

---

The Critical Piece: BlobCheckpointStore

Here’s what many developers miss — without proper checkpoint management, your Event Hub consumers will fail at scale.

The Problem

// ❌ WRONG - No checkpoint store
this.consumer = new EventHubConsumerClient(
  consumerGroup,
  connectionString,
  eventHubName
  // Missing checkpointStore!
);

Without BlobCheckpointStore, you get:

1. No partition ownership coordination — Pods don’t know who owns which partition
2. In-memory checkpoints lost — On pod restart, events are missed or reprocessed
3. No rebalancing — When pods scale, partitions don’t redistribute
4. Starvation — Some pods may never receive events

The Solution

import { BlobCheckpointStore } from '@azure/eventhubs-checkpointstore-blob';
import { ContainerClient } from '@azure/storage-blob';

private async initializeConsumer() {
  // Create checkpoint store backed by Azure Blob Storage
  const containerClient = new ContainerClient(
    process.env.AZURE_STORAGE_CONNECTION_STRING,
    'eventhub-checkpoints'
  );
  const checkpointStore = new BlobCheckpointStore(containerClient);

  // Create consumer WITH checkpoint store
  this.consumer = new EventHubConsumerClient(
    consumerGroup,
    connectionString,
    eventHubName,
    checkpointStore  // ✅ This enables partition management
  );

  this.subscription = this.consumer.subscribe({
    processEvents: this.processEvents.bind(this),
    processError: this.processError.bind(this),
  });
}

How BlobCheckpointStore Works

It stores two critical pieces of information in Azure Blob Storage:

Azure Blob Storage
└── eventhub-checkpoints/
    ├── ownership/
    │   ├── partition-0 → "pod-1-uuid" (expires in 60s)
    │   ├── partition-1 → "pod-2-uuid"
    │   └── partition-2 → "pod-3-uuid"
    └── checkpoint/
        ├── partition-0 → {offset: 1234, sequenceNumber: 567}
        ├── partition-1 → {offset: 2345, sequenceNumber: 678}
        └── partition-2 → {offset: 3456, sequenceNumber: 789}

Automatic Rebalancing:

• Pod 3 dies → ownership expires in 60s → Pod 1 or 2 claims partition-2
• Pod 4 starts → steals a partition from the busiest pod
• Scale down → remaining pods claim orphaned partitions

---

Connection Management: The 94.7% Hit Rate Secret

With multi-tenant architecture, each tenant has their own database. Creating a new connection for every event would be catastrophic at scale.

LRU-Cached Connection Pooling

class TenantConnectionManager {
  private cache: LRUCache<string, DataSource>;
  private refCounts: Map<string, number>;

  constructor() {
    this.cache = new LRUCache({
      max: 20,  // Keep 20 most recently used connections
      ttl: 600000,  // 10-minute idle timeout
      dispose: (ds) => ds.destroy()  // Clean up on eviction
    });
  }

  async getConnection(tenantId: string): Promise<DataSource> {
    const cached = this.cache.get(tenantId);
    if (cached) {
      this.refCounts.set(tenantId, (this.refCounts.get(tenantId) || 0) + 1);
      return cached;  // ✅ Cache hit
    }

    // Create new connection
    const dataSource = await this.createDataSource(tenantId);
    this.cache.set(tenantId, dataSource);
    return dataSource;
  }

  releaseConnection(tenantId: string) {
    const refs = (this.refCounts.get(tenantId) || 1) - 1;
    this.refCounts.set(tenantId, refs);
    // Connection stays in cache for future requests
  }
}

Results

Metric	Before	After
Cache Hit Rate	0%	94.7%
Memory per Pod	320 MB	110 MB
Connection Setup Time	Every request	Only on cache miss

---

Eliminating N+1 Queries: 100x Performance Boost

One of the biggest performance wins came from fixing N+1 query patterns.

The Anti-Pattern

// ❌ SLOW - N+1 queries
for (const customer of customers) {
  const pastTrigger = await checkIfSent(customer.id);  // 1000 queries!
}

The Fix

// ✅ FAST - Single batch query
const pastTriggers = await queryRunner.manager.find(TriggerLogEntity, {
  where: { 
    customer_id: In(customerIds), 
    trigger_id 
  }
});

const sentSet = new Set(pastTriggers.map(t => t.customer_id));

for (const customer of customers) {
  if (sentSet.has(customer.id)) continue;  // O(1) lookup
  await sendNotification(customer);
}

Impact

• 1,000 customers: 1,000 queries → 1 query
• Latency: 3,000ms → 30ms (100x faster)

---

Resilience Patterns

1. Duplicate Detection with Redis

async processEvent(event: EventData) {
  const lockKey = `event:${event.systemProperties.sequenceNumber}`;
  
  // Try to acquire lock (NX = only if not exists)
  const acquired = await redis.set(lockKey, '1', 'NX', 'EX', 300);
  
  if (!acquired) {
    logger.debug('Event already being processed, skipping');
    return;  // Another pod is handling this
  }

  try {
    await this.handleEvent(event);
  } finally {
    // Lock expires automatically after 5 minutes
  }
}

2. Health Check with Auto-Reconnect

private lastEventTime: Date;

@Cron('*/2 * * * *')  // Every 2 minutes
async healthCheck() {
  const minutesSinceLastEvent = 
    (Date.now() - this.lastEventTime.getTime()) / 60000;

  if (minutesSinceLastEvent > 5) {
    logger.warn('Consumer stale, reconnecting...');
    await this.consumer.close();
    await this.initializeConsumer();
  }
}

3. Exponential Backoff Retries

async processWithRetry(event: any, attempt = 0) {
  const DELAYS = [1000, 5000, 15000];  // 1s, 5s, 15s
  
  try {
    await this.processEvent(event);
  } catch (error) {
    if (attempt < 3 && this.isRetryable(error)) {
      await this.delay(DELAYS[attempt]);
      return this.processWithRetry(event, attempt + 1);
    }
    
    // Send to Dead Letter Queue after 3 failures
    await this.sendToDeadLetterQueue(event, error);
  }
}

---

Kubernetes HPA Configuration

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: event-processor
spec:
  minReplicas: 10
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Pods
        value: 2
        periodSeconds: 60

---

Final Architecture

flowchart TB
    subgraph Ingestion["INGESTION LAYER"]
        EH["Azure Event Hub<br/>32 Partitions | 32K evt/sec"]
        BCS["BlobCheckpointStore<br/>Partition ownership + Checkpoints"]
    end
    
    subgraph Processing["PROCESSING LAYER"]
        P1["Pod 1<br/>333 e/s"]
        P2["Pod 2<br/>333 e/s"]
        P3["Pod 3<br/>333 e/s"]
        PN["Pod N<br/>333 e/s"]
        CM["Connection Manager<br/>LRU Cache | 94.7% hit rate"]
    end
    
    subgraph DataLayer["DATA LAYER"]
        PG[(PostgreSQL<br/>Primary)]
        MY[(MySQL<br/>POS Tenants)]
        RD[(Redis<br/>Cache/DLQ)]
    end
    
    EH --> BCS
    BCS --> P1
    BCS --> P2
    BCS --> P3
    BCS --> PN
    
    P1 --> CM
    P2 --> CM
    P3 --> CM
    PN --> CM
    
    CM --> PG
    CM --> MY
    CM --> RD
    
    style EH fill:#4ecdc4,stroke:#333,stroke-width:2px
    style CM fill:#64ffda,stroke:#333,stroke-width:2px

---

Key Takeaways

1. BlobCheckpointStore is essential — Without it, scaling is impossible
2. Cache your connections — 94.7% hit rate saves massive resources
3. Batch your queries — N+1 patterns kill performance
4. Plan for failures — Retries, DLQ, and health checks are non-negotiable
5. Monitor everything — You can’t optimize what you can’t measure

Performance Results

Metric	Target	Achieved
Throughput	3,333 evt/s	✅ 3,333 evt/s
Peak Capacity	10,000 evt/s	✅ 32,000 evt/s
Latency (p50)	<100ms	✅ 50ms
Latency (p99)	<1s	✅ 500ms
Error Rate	<1%	✅ <0.1%
Availability	99.9%	✅ 99.9%

---

Building high-scale systems requires careful attention to these details. The difference between a system that works and one that works at scale is often in the implementation patterns you choose. See also: PostgreSQL Materialized Views for Segmentation, LRU Connection Pooling, and CQRS in Practice.

Questions? Connect with me on LinkedIn.