Apache Pulsar Producer with Python Client

Apache Pulsar Producer with Python Client

Problem Statement

Modern distributed systems need reliable, high-throughput messaging to decouple services, buffer load spikes, and enable event-driven architectures. Apache Pulsar provides multi-tenancy, geo-replication, and durable message storage, but writing a robust producer that handles batching, retries, backpressure, and graceful shutdown correctly is non-trivial.

This script provides a production-ready Pulsar producer CLI that publishes JSON messages to a topic with configurable batching, compression, and delivery guarantees. It's useful as a standalone tool for load testing, as a reference implementation for services, and as a base for ETL pipelines that need to push events into Pulsar from cron jobs or scripts.

How It Works

The script creates a pulsar.Client with a bounded producer configuration, then sends a configurable number of messages (either generated or read from stdin) to the target topic. Messages are serialized as JSON, partitioned via a key (so related messages land on the same partition), and batched to maximize throughput. The producer uses SendCallback-style async sends via send_async with a semaphore to bound in-flight messages — this is the standard backpressure pattern for the Pulsar Python client, which is synchronous by default.

Key design decisions: (1) a Semaphore caps concurrent in-flight sends, preventing memory blowup when the broker is slow; (2) batching_enabled is on with a configurable delay so small messages get coalesced; (3) a finally block flushes and closes the producer so no messages are lost on Ctrl-C; (4) configuration comes from env vars with CLI overrides, following the 12-factor pattern.

Full Script

#!/usr/bin/env python3
"""
Apache Pulsar Producer with Python Client

Publishes JSON messages to a Pulsar topic with configurable batching,
compression, and backpressure. Useful as a CLI tool for load testing,
event injection, or as a reference for embedding in services.
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import signal
import sys
import threading
import time
import uuid
from typing import Any

import pulsar
from pulsar import Client, Producer

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults, overridable via CLI)
# ---------------------------------------------------------------------------
PULSAR_URL = os.environ.get("PULSAR_URL", "pulsar://localhost:6650")
PULSAR_TOPIC = os.environ.get(
    "PULSAR_TOPIC", "persistent://public/default/orders"
)
PULSAR_AUTH_TOKEN = os.environ.get("PULSAR_AUTH_TOKEN")  # optional JWT
PULSAR_TLS_TRUST_CERTS = os.environ.get("PULSAR_TLS_TRUST_CERTS")  # optional
BATCH_MAX_MESSAGES = int(os.environ.get("PULSAR_BATCH_MAX_MESSAGES", "1000"))
BATCH_MAX_DELAY_MS = int(os.environ.get("PULSAR_BATCH_MAX_DELAY_MS", "10"))
MAX_IN_FLIGHT = int(os.environ.get("PULSAR_MAX_IN_FLIGHT", "1000"))
SEND_TIMEOUT_S = int(os.environ.get("PULSAR_SEND_TIMEOUT_S", "30"))

logging.basicConfig(
    level=os.environ.get("LOG_LEVEL", "INFO"),
    format="%(asctime)s %(levelname)-7s %(message)s",
)
log = logging.getLogger("pulsar-producer")


class PulsarProducer:
    """Thin wrapper around pulsar.Producer with backpressure + shutdown."""

    def __init__(self, client: Client, topic: str) -> None:
        self._client = client
        self._topic = topic
        self._producer: Producer = client.create_producer(
            topic,
            # Batching: coalesce small messages into one broker round-trip.
            batching_enabled=True,
            batching_max_messages=BATCH_MAX_MESSAGES,
            batching_max_publish_delay_ms=BATCH_MAX_DELAY_MS,
            # Compression reduces network + storage cost.
            compression_type=pulsar.CompressionType.LZ4,
            # Block (rather than error) when the broker's send queue is full.
            block_if_queue_full=True,
            # Bound the number of unacked messages in flight.
            max_pending_messages=MAX_IN_FLIGHT,
            # Wait for the broker's ack — no silent data loss.
            send_timeout_millis=SEND_TIMEOUT_S * 1000,
        )
        # Semaphore enforces backpressure on the *client* side too.
        self._sem = threading.Semaphore(MAX_IN_FLIGHT)
        self._lock = threading.Lock()
        self._sent = 0
        self._failed = 0
        self._closed = False

    def send(self, payload: dict[str, Any], key: str | None = None) -> None:
        """Send one JSON message; blocks if in-flight limit reached."""
        if self._closed:
            raise RuntimeError("producer already closed")

        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
        self._sem.acquire()
        try:
            self._producer.send_async(
                body,
                partition_key=key,
                callback=self._on_ack,
            )
        except Exception:
            # Release the slot if send_async itself raised synchronously.
            self._sem.release()
            raise

    def _on_ack(self, result: pulsar.Result, msg_id: pulsar.MessageId) -> None:
        """Callback invoked by the Pulsar client thread on ack/error."""
        with self._lock:
            if result == pulsar.Result.Ok:
                self._sent += 1
            else:
                self._failed += 1
                log.error("send failed: result=%s msg_id=%s", result, msg_id)
        self._sem.release()

    def flush_and_close(self) -> None:
        """Block until all in-flight messages are acked, then close."""
        if self._closed:
            return
        self._closed = True
        # Drain: wait for every acquired permit to be released.
        for _ in range(MAX_IN_FLIGHT):
            self._sem.acquire()
        self._producer.flush()
        self._producer.close()
        log.info(
            "producer closed: sent=%d failed=%d", self._sent, self._failed
        )

    @property
    def stats(self) -> tuple[int, int]:
        return self._sent, self._failed


def build_client() -> Client:
    """Build a Pulsar client honoring auth + TLS env vars."""
    kwargs: dict[str, Any] = {
        "operation_timeout_seconds": SEND_TIMEOUT_S,
        "connection_timeout_ms": 10_000,
    }
    if PULSAR_AUTH_TOKEN:
        kwargs["authentication"] = pulsar.AuthenticationToken(
            PULSAR_AUTH_TOKEN
        )
    if PULSAR_TLS_TRUST_CERTS:
        kwargs["tls_trust_certs_file_path"] = PULSAR_TLS_TRUST_CERTS
    return Client(PULSAR_URL, **kwargs)


def make_payload(seq: int, source: str) -> dict[str, Any]:
    """Build a simple, realistic order event payload."""
    return {
        "event_id": str(uuid.uuid4()),
        "sequence": seq,
        "source": source,
        "created_at": time.time(),
        "amount": round(10 + (seq % 90) * 1.37, 2),
        "currency": "USD",
    }


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Pulsar JSON producer")
    p.add_argument("--url", default=PULSAR_URL, help="Pulsar service URL")
    p.add_argument("--topic", default=PULSAR_TOPIC, help="Topic name")
    p.add_argument(
        "--count", type=int, default=100, help="Messages to send (0=stdin)"
    )
    p.add_argument(
        "--rate", type=float, default=0.0, help="Messages/sec (0=unlimited)"
    )
    p.add_argument(
        "--key-field",
        default="event_id",
        help="Payload field used as partition key",
    )
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    client = build_client()
    producer = PulsarProducer(client, args.topic)

    # Ensure Ctrl-C flushes pending messages instead of dropping them.
    stop = threading.Event()

    def _handle_sigint(signum: int, frame: Any) -> None:
        log.warning("signal %d received — shutting down", signum)
        stop.set()

    signal.signal(signal.SIGINT, _handle_sigint)
    signal.signal(signal.SIGTERM, _handle_sigint)

    interval = 1.0 / args.rate if args.rate > 0 else 0.0
    seq = 0
    try:
        if args.count > 0:
            for seq in range(args.count):
                if stop.is_set():
                    break
                payload = make_payload(seq, "cli")
                key = str(payload.get(args.key_field, "")) or None
                producer.send(payload, key=key)
                if interval:
                    time.sleep(interval)
        else:
            # Stream JSON objects, one per line, from stdin.
            for line in sys.stdin:
                if stop.is_set():
                    break
                line = line.strip()
                if not line:
                    continue
                payload = json.loads(line)
                key = str(payload.get(args.key_field, "")) or None
                producer.send(payload, key=key)
                seq += 1
    except json.JSONDecodeError as exc:
        log.error("invalid JSON on stdin: %s", exc)
        return 2
    except pulsar.PulsarException as exc:
        log.error("pulsar error: %s", exc)
        return 3
    finally:
        producer.flush_and_close()
        client.close()

    sent, failed = producer.stats
    log.info("done: attempted=%d sent=%d failed=%d", seq, sent, failed)
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())

Configuration

Usage Examples

1. Send 500 generated orders at 200 msg/s:

$ python3 pulsar_producer.py --count 500 --rate 200
2024-05-12 10:14:02 INFO    producer closed: sent=500 failed=0
2024-05-12 10:14:02 INFO    done: attempted=500 sent=500 failed=0

2. Stream JSON events from a file (one per line):

$ cat orders.jsonl | python3 pulsar_producer.py --count 0 --topic persistent://public/default/orders
2024-05-12 10:15:11 INFO    producer closed: sent=42 failed=0
2024-05-12 10:15:11 INFO    done: attempted=42 sent=42 failed=0

3. Send to a TLS + token-authenticated cluster:

$ PULSAR_AUTH_TOKEN="$(cat token.jwt)" \
  PULSAR_TLS_TRUST_CERTS=/etc/pulsar/ca.pem \
  python3 pulsar_producer.py \
    --url pulsar+ssl://broker.example.com:6651 \
    --topic persistent://acme/orders/created \
    --count 10000 --rate 0

Error Handling

• Broker unreachable / auth failure — pulsar.Client(...) raises pulsar.ConnectError; the script exits non-zero before sending anything.
• Invalid JSON on stdin — caught as json.JSONDecodeError, logged, and returns exit code 2.
• Send failures — per-message errors arrive via _on_ack with a non-Ok result. They're counted in failed, logged, and cause exit code 1 so cron/CI can detect partial failure.
• Backpressure — the client-side Semaphore(MAX_IN_FLIGHT) blocks send() when the broker is slow, preventing unbounded memory growth; block_if_queue_full=True does the same on the broker side.
• Ctrl-C / SIGTERM — the signal handler sets a stop event; the finally block calls flush_and_close(), which drains in-flight acks before closing so no messages are silently dropped.
• Pulsar exceptions — pulsar.PulsarException is caught in main() and returns exit code 3.

Extending Ideas

• Schema Registry — replace json.dumps with a pulsar.schema.JsonSchema producer so the broker enforces Avro/JSON schema evolution across producers.
• Prometheus metrics — expose sent, failed, and queue depth via prometheus_client on a /metrics endpoint for observability in Kubernetes.
• Async producer with asyncio — swap the threaded client for pulsar.asyncio.Client to integrate natively with async services and use asyncio.Semaphore for backpressure.
• Dead-letter topic — publish messages that exceed a retry threshold to persistent://public/default/orders-dlq for later inspection instead of just counting failures.