Pulsar Python Client

Warnings

• breaking Pulsar Python client versions are generally aligned with Apache Pulsar Broker versions. Using a client version (e.g., 3.x) with an older broker (e.g., 2.x) or vice-versa might lead to unexpected behavior, missing features, or connection failures. Always ensure compatibility between client and broker versions.
  Fix: Refer to the official Apache Pulsar documentation for the recommended client-broker compatibility matrix. Upgrade your client and/or broker to compatible versions.
• gotcha Failing to explicitly call `.close()` on `pulsar.Client`, `Producer`, and `Consumer` objects can lead to resource leaks (e.g., open connections, file descriptors) and connection issues in long-running applications.
  Fix: Always ensure `.close()` is called on these objects when they are no longer needed, typically in `finally` blocks or by using `with` statements if available (though not directly supported by current `pulsar-client` objects in a context manager way for `close()`).
• gotcha Correctly configuring authentication (e.g., JWT tokens) and TLS for secure connections is a common source of initial setup errors. Incorrect parameters can lead to connection refused errors or authentication failures.
  Fix: Carefully review the documentation for `pulsar.Client` constructor parameters like `authentication`, `tls_enable`, `tls_trust_certs_file`, `tls_allow_insecure_connection`, etc. Ensure paths to certificate files are correct and tokens are valid. Set `tls_enable=True` for TLS connections.

Install

• pip install pulsar-client Install stable version

Imports

• Client

  from pulsar import Client

• Producer

  from pulsar import Producer

• Consumer

  from pulsar import Consumer

• MessageId

  from pulsar import MessageId

• Schema

  from pulsar import Schema

  Used for structured message types like Avro or JSON schemas.
• ConsumerType

  from pulsar import ConsumerType

  Defines how messages are delivered to consumers within a subscription.

Quickstart

This quickstart demonstrates how to produce and consume messages using the `pulsar-client` library. It assumes a Pulsar broker is running and accessible (e.g., via `pulsar://localhost:6650`). The script first acts as a producer, sending three messages to a topic, then switches to a consumer, subscribing to the same topic and receiving those messages. Crucially, it shows proper resource management by calling `.close()` on clients, producers, and consumers. The Pulsar broker URL can be configured via the `PULSAR_BROKER_URL` environment variable.

import pulsar
import time
import os

# --- Configuration ---
# Ensure a Pulsar broker is running, e.g., locally via Docker:
# docker run -it -p 6650:6650 -p 8080:8080 apachepulsar/pulsar:latest bin/pulsar standalone
PULSAR_BROKER_URL = os.environ.get('PULSAR_BROKER_URL', 'pulsar://localhost:6650')
TOPIC_NAME = 'persistent://public/default/my-python-topic-qs'
SUBSCRIPTION_NAME = 'my-python-subscription-qs'

print(f"Connecting to Pulsar at: {PULSAR_BROKER_URL}")

# --- Producer ---
try:
    print("\n--- Starting Producer ---")
    client_producer = pulsar.Client(PULSAR_BROKER_URL)
    producer = client_producer.create_producer(TOPIC_NAME)

    for i in range(3):
        message_data = f"hello-pulsar-message-{i}".encode('utf-8')
        producer.send(message_data)
        print(f"Producer sent: {message_data.decode()}")
        time.sleep(0.1) # Small delay

    producer.close()
    client_producer.close()
    print("--- Producer finished. ---")
except Exception as e:
    print(f"Producer error: {e}")

time.sleep(1) # Give a moment for messages to be available in the topic

# --- Consumer ---
try:
    print("\n--- Starting Consumer ---")
    client_consumer = pulsar.Client(PULSAR_BROKER_URL)
    consumer = client_consumer.subscribe(TOPIC_NAME, SUBSCRIPTION_NAME, consumer_type=pulsar.ConsumerType.Shared)

    print("Consumer waiting for messages...")
    received_count = 0
    # We expect 3 messages from the producer
    while received_count < 3:
        try:
            msg = consumer.receive(timeout_millis=3000) # Wait up to 3 seconds
            if msg:
                print(f"Consumer received: '{msg.data().decode('utf-8')}' (ID: {msg.message_id()})")
                consumer.acknowledge(msg)
                received_count += 1
            else:
                print("Consumer timed out waiting for message. Retrying...")
        except Exception as msg_err:
            print(f"Error receiving individual message: {msg_err}")
            break # Exit on error
    
    consumer.close()
    client_consumer.close()
    print("---
 Consumer finished. ---")
except Exception as e:
    print(f"Consumer error: {e}")

print("\nQuickstart demonstration complete.")