Kafka vs Pub/Sub

Kafka stores messages as an ordered, partitioned, immutable log. Each topic is divided into partitions (typically 3 to 100+), each partition maintained on one broker (leader) with replicas on others. Producers route messages to partitions via a partition key: the same order ID always lands on the same partition, guaranteeing ordering per key. Each consumer in a group owns a set of partitions and tracks its own offset (position in the log). Consumer groups are independent: Group A at offset 84,000 and Group B at offset 12 are reading the same partition simultaneously.

Replay is a first-class operation. Seek any consumer group to any offset or timestamp within the retention window (typically 7 days). Replaying the Bronze ingestion pipeline after a bad schema deployment takes two commands and does not affect any other consumer group. The Schema Registry (Confluent SR or Apicurio) enforces Avro or Protobuf compatibility at publish time: breaking changes are rejected before they reach the topic.

Pub/Sub stores messages in a topic and delivers a copy to each subscription independently. There are no partitions or offsets: the service manages delivery internally. Pull subscribers call the API to receive messages; push subscriptions deliver to an HTTP endpoint. Every message has an acknowledgement (ack) deadline. If the subscriber does not ack within the window (default 10s, configurable to 600s), Pub/Sub redelivers. At-least-once delivery is guaranteed; exactly-once requires idempotent consumers.

After max_delivery_attempts failed attempts (configurable per subscription), Pub/Sub routes the message to a dead letter topic (DLT) automatically. Replay via Pub/Sub Seek lets you rewind all subscriptions on a topic to a snapshot or timestamp, but it cannot replay one subscription independently of others. Pub/Sub Schemas provide basic Avro or Protobuf validation at publish time, but do not enforce compatibility evolution rules the way Confluent Schema Registry does.

Kafka producer with Avro and Confluent Schema Registry

Partition key is the order_id. Same order always routes to same partition. acks=all means the leader waits for all in-sync replicas before acknowledging.

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField

ORDER_SCHEMA = """
{
  "type": "record",
  "name": "OrderPlaced",
  "namespace": "com.example.orders",
  "fields": [
    {"name": "order_id",      "type": "string"},
    {"name": "customer_id",   "type": "string"},
    {"name": "amount_cents",  "type": "long"},
    {"name": "placed_at_ms",  "type": {"type": "long", "logicalType": "timestamp-millis"}}
  ]
}
"""

sr_client = SchemaRegistryClient({"url": "https://sr.example.com"})
avro_serializer = AvroSerializer(sr_client, ORDER_SCHEMA)

producer = Producer({
    "bootstrap.servers": "kafka.example.com:9092",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "API_KEY",
    "sasl.password": "API_SECRET",
    # Reliability: wait for leader + all in-sync replicas
    "acks": "all",
    "retries": 3,
})

def delivery_report(err, msg):
    if err:
        raise RuntimeError(f"Delivery failed: {err}")

event = {
    "order_id": "ord-12345",
    "customer_id": "cust-67890",
    "amount_cents": 4999,
    "placed_at_ms": 1710000000000,
}

producer.produce(
    topic="orders.placed.v1",
    key="ord-12345",           # partition key: same order_id always same partition
    value=avro_serializer(
        event,
        SerializationContext("orders.placed.v1", MessageField.VALUE)
    ),
    on_delivery=delivery_report,
)
producer.flush()

Kafka consumer with consumer group and manual offset commit

enable.auto.commit=False is non-negotiable for at-least-once delivery. Auto-commit can acknowledge messages before they have been processed successfully.

from confluent_kafka import Consumer, KafkaException
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
import logging

sr_client = SchemaRegistryClient({"url": "https://sr.example.com"})
avro_deserializer = AvroDeserializer(sr_client)

consumer = Consumer({
    "bootstrap.servers": "kafka.example.com:9092",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.username": "API_KEY",
    "sasl.password": "API_SECRET",
    "group.id": "orders-ingestion-pipeline",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,   # manual commit: only after successful processing
    "max.poll.interval.ms": 300000,
})
consumer.subscribe(["orders.placed.v1"])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())

        event = avro_deserializer(
            msg.value(),
            SerializationContext(msg.topic(), MessageField.VALUE),
        )

        try:
            process_order(event)
            # Commit only after successful write to Bronze. Guarantees at-least-once delivery.
            consumer.commit(asynchronous=False)
        except Exception as exc:
            # Do NOT commit. Message will be redelivered on next poll.
            logging.error("Processing failed, skipping commit", exc_info=exc)

except KeyboardInterrupt:
    pass
finally:
    consumer.close()

Pub/Sub publisher with message attributes

Message attributes carry routing and tracing metadata without polluting the payload schema. future.result() blocks until Pub/Sub acknowledges storage, not delivery.

from google.cloud import pubsub_v1
import json

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("my-gcp-project", "orders.placed.v1")

def publish_order_placed(
    order_id: str,
    customer_id: str,
    amount_cents: int,
    placed_at_ms: int,
) -> str:
    payload = json.dumps({
        "order_id": order_id,
        "customer_id": customer_id,
        "amount_cents": amount_cents,
        "placed_at_ms": placed_at_ms,
    }).encode("utf-8")

    future = publisher.publish(
        topic_path,
        data=payload,
        # Message attributes used for filtering and tracing
        domain="orders",
        event_type="placed",
        schema_version="v1",
        correlation_id=order_id,
    )
    # .result() blocks until Pub/Sub acknowledges receipt
    message_id = future.result()
    return message_id

Pub/Sub pull subscriber with dead letter topic handling

The dead letter topic is configured on the subscription in Terraform, not in code. The subscriber only needs to distinguish permanent vs. transient failures when deciding whether to ack or nack.

from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError
import json
import logging

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(
    "my-gcp-project",
    "orders.placed.v1-ingestion-pipeline",  # subscription, not topic
)

def handle_message(message: pubsub_v1.types.PubsubMessage) -> None:
    try:
        payload = json.loads(message.data.decode("utf-8"))
        process_order(payload)

        # Ack removes the message from this subscription permanently
        message.ack()

    except json.JSONDecodeError as exc:
        # Permanent failure: bad payload. Ack to prevent infinite redelivery.
        # Route to dead letter via message attribute or separate pipeline.
        logging.error("Unparseable message, acking to DLT path", exc_info=exc)
        message.ack()

    except Exception as exc:
        # Transient failure: nack triggers redeliver after ack deadline.
        # After max_delivery_attempts the subscription routes to the dead letter topic.
        logging.warning("Transient failure, nacking for redelivery", exc_info=exc)
        message.nack()

streaming_pull_future = subscriber.subscribe(
    subscription_path,
    callback=handle_message,
)

with subscriber:
    try:
        streaming_pull_future.result()
    except Exception:
        streaming_pull_future.cancel()
        streaming_pull_future.result()

Kafka Connect for CDC pipelines

Kafka Connect runs source connectors that pull from external systems and publish to Kafka topics. For CDC (Change Data Capture), the Debezium connector tails the database binlog (MySQL binary log, Postgres WAL, Oracle LogMiner) and publishes row-level changes as Kafka messages. Single Message Transforms (SMTs) let you rename fields, filter rows, or add metadata before the message reaches the topic, without writing a custom Kafka Streams job.

Schema registry comparison: Confluent SR vs Apicurio

Both support Avro, Protobuf, and JSON Schema. Both enforce compatibility modes (BACKWARD, FORWARD, FULL). The differences are operational: Confluent Schema Registry is a single binary with a REST API; it is simple to run and well-documented, but it is part of the Confluent Platform licensing model when used with Confluent Cloud. Apicurio Registry is an open-source alternative that also supports OpenAPI and AsyncAPI artifacts, making it a better fit if you need a central contract registry beyond just Kafka schemas. Use Confluent SR if you are already on Confluent Cloud, where it is bundled. Use Apicurio if you want a schema registry independent of Confluent licensing or if you need multi-format artifact storage.

Kafka consumer group lag monitoring

Consumer lag is the difference between the latest produced offset and the last committed offset for a consumer group on a partition. It is the primary freshness signal for Kafka-backed pipelines: a lag of 50,000 messages at 10,000 msg/sec means you are 5 seconds behind. At 1,000 msg/sec on a low-throughput topic, the same lag means 50 seconds behind.

Pub/Sub dead letter topics: configuration and behaviour

A dead letter topic (DLT) is configured at the subscription level, not the topic level. Each subscription can have a different DLT and a different max_delivery_attempts value (minimum 5, maximum 100). When a message is nacked or the ack deadline expires max_delivery_attempts times, Pub/Sub moves it to the DLT automatically with a CloudPubSubDeadLetterPolicy message attribute indicating the original subscription and delivery attempt count.

The DLT is itself a Pub/Sub topic. Create a separate subscription on the DLT to inspect and replay failed messages. Pub/Sub does not automatically retry DLT messages: you own the replay logic. A common pattern is a Cloud Function or small Dataflow job subscribed to the DLT that logs the failure, sends an alert, and republishes to the original topic after human review or automated correction.

Kafka anti-patterns

• !
  Using Kafka as a database. Setting retention to "infinite" and querying Kafka for current state via consumer seeks. Kafka is a log, not a queryable store. Current state belongs in a database or data warehouse. Use Kafka Streams or ksqlDB materialised views if you need stateful queries, but accept the operational cost.
• !
  Too many small topics. Creating a separate topic for every microservice event type with 3 partitions each. 200 microservices × 5 event types × 3 partitions = 3,000 partitions on a 4-broker cluster. ZooKeeper (or KRaft) controller election time scales with partition count. Keep topics coarse-grained at the domain level. Use message type headers or schema fields to distinguish event types within a topic.
• !
  No partition key discipline. Using null partition keys (round-robin assignment) for events that require ordering, then wondering why the order of events for the same entity is inconsistent. Establish and document the canonical partition key for each topic before the first message is published.
• !
  No schema enforcement at produce time. Letting producers publish raw JSON without schema registration. A producer deploys a change, renames a field, and every downstream consumer crashes silently. The schema registry check at produce time is the only reliable gate.

Pub/Sub anti-patterns

• !
  Assuming exactly-once delivery. Pub/Sub guarantees at-least-once. Under normal conditions most messages are delivered once, but during redelivery windows, restarts, or subscription resets, duplicates occur. Every consumer must be idempotent. Using a message ID or a business key as an upsert key in BigQuery or Spanner is the correct pattern.
• !
  Ignoring ordering on ordered subscriptions. Enabling message ordering keys without understanding the performance trade-off: ordered delivery serialises processing per key. One slow ack for key "order-123" blocks all subsequent messages with the same key. For high-cardinality keys (order IDs) the impact is minimal. For low-cardinality keys (region codes) it can stall an entire subscription.
• !
  Using Pub/Sub for high-fan-out CDC. Routing full-table CDC from a high-write-volume Oracle table through Pub/Sub to 10+ downstream subscriptions. At 5 TB/day ingress with 10 subscriptions, Pub/Sub delivery costs alone exceed $5,000/month. Kafka with 10 consumer groups on the same data costs roughly the same as 1 consumer group. CDC fan-out is a primary use case where Kafka's cost model wins.