Event Hubs Push vs Pull: How Consumers Work

Push vs pull: the fundamental difference

In a push model, the messaging service sends events to your consumer as they arrive. Your application must be ready to receive them at all times. If your application is busy or slow, the service buffers messages and eventually times out or drops events. Examples: HTTP webhooks, Azure Service Bus push subscriptions, Azure Event Grid.

In a pull model, your consumer periodically asks the messaging service “give me the next batch of events.” The consumer controls the rate. If it is busy, it simply does not ask for more events until it is ready. The service holds all events in the partition log until the consumer retrieves them (or until retention expires). Examples: Azure Event Hubs, Apache Kafka, Amazon Kinesis.

How the pull model works in practice

When your Event Hubs consumer calls receive_batch() or receive(), it sends a request to the Event Hubs broker saying “give me up to N events from partition P, starting at offset X.” The broker responds with the next available events. The consumer processes them, then calls receive again for the next batch.

This loop runs continuously for as long as the consumer is active:

import asyncio
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
import json
import time

async def process_event_batch(partition_context, events):
    """
    This function is called each time a batch of events is available.
    The consumer pulls batches at its own pace.
    """
    start_time = time.time()
    processed = 0

    for event in events:
        payload = json.loads(event.body_as_str())

        # Simulate processing (database write, ML inference, etc.)
        await process_single_event(payload)
        processed += 1

    elapsed = time.time() - start_time
    rate = processed / elapsed if elapsed > 0 else 0

    print(
        f"Partition {partition_context.partition_id}: "
        f"processed {processed} events in {elapsed:.2f}s ({rate:.0f} events/sec)"
    )

    # Checkpoint after the whole batch is processed
    # If we crash here, we restart from the beginning of this batch (at-least-once)
    await partition_context.update_checkpoint()


async def process_single_event(payload: dict):
    """Simulate event processing with some I/O"""
    # In real code: write to DB, call an API, update a cache, etc.
    await asyncio.sleep(0.001)  # 1ms simulated processing per event


async def main():
    checkpoint_store = BlobCheckpointStore.from_connection_string(
        conn_str="DefaultEndpointsProtocol=https;AccountName=mystorageacct;...",
        container_name="checkpoints"
    )

    client = EventHubConsumerClient.from_connection_string(
        conn_str="Endpoint=sb://eh-ns-mycompany.servicebus.windows.net/;...",
        consumer_group="myapp-cg",
        eventhub_name="orders",
        checkpoint_store=checkpoint_store
    )

    print("Starting consumer — pulling events from Event Hubs...")
    async with client:
        # receive_batch calls process_event_batch whenever events are available
        # max_batch_size: pull up to 200 events per batch
        # max_wait_time: wait up to 5 seconds for events before calling back with empty batch
        await client.receive_batch(
            on_event_batch=process_event_batch,
            max_batch_size=200,
            max_wait_time=5
        )

asyncio.run(main())

Notice that the consumer controls the batch size (max_batch_size=200). If your processing function takes 2 seconds per batch and you pull 200 events per batch, you process 100 events per second. If your source is producing 1,000 events per second, you will fall behind — the unconsumed events simply stay in the partition log until you catch up.

Backpressure: the pull model’s natural protection

Backpressure is the mechanism by which a slow consumer protects itself from being overwhelmed. In the pull model, backpressure is automatic — if your processing is slow, your polling slows down, and you simply fall behind in the partition log. No configuration needed.

The lag (how far behind the consumer is from the latest event) is measurable. You can monitor it using the Event Hubs metrics in Azure Monitor:

# Check consumer lag for a specific consumer group using Azure CLI
# "Consumer Lag" metric shows how many unprocessed events remain per partition
az monitor metrics list \
  --resource "/subscriptions/{sub}/resourceGroups/rg-streaming/providers/Microsoft.EventHub/namespaces/eh-ns-mycompany" \
  --metric "IncomingMessages,OutgoingMessages" \
  --interval PT1M \
  --aggregation Total \
  --output table

If lag is growing persistently, you have three options:

1. Add more consumer instances (horizontal scaling — each handles a subset of partitions)
2. Increase the partition count of the event hub (requires recreating the hub on standard tier)
3. Optimise your processing logic to handle events faster

Seeking and replaying events

One of the most powerful features of the pull model is the ability to seek — to position your consumer at any point in the partition history and reprocess events from that point. This is impossible with push-based systems where events are deleted after delivery.

from azure.eventhub import EventHubConsumerClient, EventPosition
from datetime import datetime, timezone, timedelta

# Scenario: a bug in your processing code corrupted data written to your database.
# The bug was introduced 6 hours ago.
# You want to replay all events from 6 hours ago to the present.

six_hours_ago = datetime.now(timezone.utc) - timedelta(hours=6)

# Set starting positions to 6 hours ago for all partitions
starting_positions = {
    "0": EventPosition(six_hours_ago),
    "1": EventPosition(six_hours_ago),
    "2": EventPosition(six_hours_ago),
    "3": EventPosition(six_hours_ago),
}

# The checkpoint store currently has offsets from the current position.
# To force a replay, either:
# Option A: delete the checkpoint blobs in storage (they will start from starting_positions)
# Option B: use a new consumer group with no existing checkpoints

client = EventHubConsumerClient.from_connection_string(
    conn_str="Endpoint=sb://...",
    consumer_group="replay-cg",   # A fresh consumer group with no checkpoints
    eventhub_name="orders"
    # No checkpoint_store — no checkpointing during replay
)

with client:
    events = client.receive(
        on_event=lambda pc, e: print(
            f"Replay: {e.sequence_number} | {e.body_as_str()[:100]}"
        ),
        starting_position=six_hours_ago,
        max_wait_time=30
    )

Kafka compatibility: using Event Hubs with Kafka clients

Event Hubs exposes a Kafka 1.0 compatible endpoint. If you already have Kafka producers or consumers written in Java, Python (confluent-kafka), or any other Kafka client library, you can point them at Event Hubs with minimal configuration changes — just update the bootstrap server and credentials.

from confluent_kafka import Consumer, KafkaError

# Connect a Kafka consumer to Event Hubs Kafka endpoint
# Event Hubs namespace acts as the Kafka broker
config = {
    "bootstrap.servers": "eh-ns-mycompany.servicebus.windows.net:9093",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": "$ConnectionString",
    "sasl.password": "Endpoint=sb://eh-ns-mycompany.servicebus.windows.net/;SharedAccessKeyName=...",
    "group.id": "kafka-consumer-group",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,   # Manual commit for reliability
}

consumer = Consumer(config)
consumer.subscribe(["orders"])   # 'orders' is the Event Hub name

print("Kafka consumer pulling from Event Hubs...")
try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print(f"Reached end of partition {msg.partition()}")
            else:
                raise Exception(msg.error())
        else:
            print(f"Received: partition={msg.partition()}, offset={msg.offset()}, value={msg.value()}")
            consumer.commit(message=msg, asynchronous=False)
finally:
    consumer.close()

This Kafka consumer reads from Event Hubs without any Event Hubs SDK. Organisations migrating from Kafka to Azure often use this compatibility layer as an intermediate step.