Apache Pulsar Core Knowledge

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: pulsar for comprehensive documentation.

Quick Start (Docker)

docker-compose.yml

services: pulsar: image: apachepulsar/pulsar:latest ports: - "6650:6650" # Broker - "8080:8080" # Admin API command: bin/pulsar standalone volumes: - pulsar_data:/pulsar/data

volumes: pulsar_data:

docker-compose up -d

Create tenant and namespace

docker exec pulsar bin/pulsar-admin tenants create my-tenant docker exec pulsar bin/pulsar-admin namespaces create my-tenant/my-namespace

Test

docker exec pulsar bin/pulsar-client produce persistent://my-tenant/my-namespace/orders -m "test" docker exec pulsar bin/pulsar-client consume persistent://my-tenant/my-namespace/orders -s "test-sub"

Core Concepts

Concept Description

Tenant Top-level isolation unit

Namespace Administrative unit within tenant

Topic Message stream (persistent/non-persistent)

Subscription Named cursor with delivery semantics

Partition Topic partitioning for parallelism

Bookie Storage layer (Apache BookKeeper)

Architecture

┌─────────────────────────────────────────────────────────────┐ │ Pulsar Cluster │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │ │ Broker 1 │ │ Broker 2 │ │ Broker 3 │ │ │ └───────────┘ └───────────┘ └───────────┘ │ │ │ │ │ │ │ ┌─────────────────────────────────────────────┐ │ │ │ BookKeeper (Bookies) │ │ │ │ ┌────────┐ ┌────────┐ ┌────────┐ │ │ │ │ │Bookie 1│ │Bookie 2│ │Bookie 3│ │ │ │ │ └────────┘ └────────┘ └────────┘ │ │ │ └─────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘

Subscription Types

Type Description Use Case

Exclusive Single consumer Ordered processing

Shared Round-robin to consumers Load balancing

Failover Active-standby High availability

Key_Shared Key-based routing Ordered per key

Producer Patterns

Node.js (pulsar-client)

import Pulsar from 'pulsar-client';

const client = new Pulsar.Client({ serviceUrl: 'pulsar://localhost:6650', operationTimeoutSeconds: 30, });

const producer = await client.createProducer({ topic: 'persistent://my-tenant/my-namespace/orders', sendTimeoutMs: 30000, batchingEnabled: true, batchingMaxMessages: 1000, batchingMaxPublishDelayMs: 10, });

// Send message const messageId = await producer.send({ data: Buffer.from(JSON.stringify(order)), properties: { 'correlation-id': correlationId, 'order-type': order.type, }, eventTimestamp: Date.now(), });

console.log(Sent: ${messageId});

// Send with key (for Key_Shared subscription) await producer.send({ data: Buffer.from(JSON.stringify(order)), partitionKey: order.customerId, });

await producer.flush(); await producer.close(); await client.close();

Java (pulsar-client)

@Configuration public class PulsarConfig { @Bean public PulsarClient pulsarClient() throws PulsarClientException { return PulsarClient.builder() .serviceUrl("pulsar://localhost:6650") .operationTimeout(30, TimeUnit.SECONDS) .build(); }

@Bean
public Producer<byte[]> orderProducer(PulsarClient client) throws PulsarClientException {
    return client.newProducer()
        .topic("persistent://my-tenant/my-namespace/orders")
        .batchingMaxMessages(1000)
        .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
        .sendTimeout(30, TimeUnit.SECONDS)
        .create();
}

}

@Service public class OrderProducer { @Autowired private Producer<byte[]> producer;

public void sendOrder(Order order) throws PulsarClientException {
    MessageId messageId = producer.newMessage()
        .value(objectMapper.writeValueAsBytes(order))
        .property("correlation-id", UUID.randomUUID().toString())
        .property("order-type", order.getType())
        .eventTime(System.currentTimeMillis())
        .key(order.getCustomerId())  // For Key_Shared
        .send();

    log.info("Sent: {}", messageId);
}

public CompletableFuture<MessageId> sendOrderAsync(Order order) {
    return producer.newMessage()
        .value(objectMapper.writeValueAsBytes(order))
        .sendAsync();
}

}

Python (pulsar-client)

import pulsar import json

client = pulsar.Client('pulsar://localhost:6650')

producer = client.create_producer( 'persistent://my-tenant/my-namespace/orders', batching_enabled=True, batching_max_messages=1000, batching_max_publish_delay_ms=10 )

Send message

message_id = producer.send( json.dumps(order).encode('utf-8'), properties={ 'correlation-id': correlation_id, 'order-type': order['type'] }, partition_key=order['customer_id'] )

print(f"Sent: {message_id}")

producer.flush() producer.close() client.close()

Go (pulsar-client-go)

package main

import ( "context" "encoding/json" "github.com/apache/pulsar-client-go/pulsar" )

func main() { client, _ := pulsar.NewClient(pulsar.ClientOptions{ URL: "pulsar://localhost:6650", OperationTimeout: 30 * time.Second, }) defer client.Close()

producer, _ := client.CreateProducer(pulsar.ProducerOptions{
    Topic:                   "persistent://my-tenant/my-namespace/orders",
    BatchingMaxMessages:     1000,
    BatchingMaxPublishDelay: 10 * time.Millisecond,
})
defer producer.Close()

body, _ := json.Marshal(order)

msgID, _ := producer.Send(context.Background(), &pulsar.ProducerMessage{
    Payload: body,
    Key:     order.CustomerID,
    Properties: map[string]string{
        "correlation-id": correlationID,
        "order-type":     order.Type,
    },
})

log.Printf("Sent: %v", msgID)

}

Consumer Patterns

Node.js

// Exclusive subscription const consumer = await client.subscribe({ topic: 'persistent://my-tenant/my-namespace/orders', subscription: 'order-processor', subscriptionType: 'Exclusive', ackTimeoutMs: 30000, });

while (true) { const msg = await consumer.receive(); try { const order = JSON.parse(msg.getData().toString()); await processOrder(order); consumer.acknowledge(msg); } catch (error) { consumer.negativeAcknowledge(msg); } }

// Shared subscription (load balanced) const sharedConsumer = await client.subscribe({ topic: 'persistent://my-tenant/my-namespace/orders', subscription: 'order-processors', subscriptionType: 'Shared', receiverQueueSize: 1000, });

// Key_Shared subscription (ordered per key) const keySharedConsumer = await client.subscribe({ topic: 'persistent://my-tenant/my-namespace/orders', subscription: 'order-processors', subscriptionType: 'KeyShared', });

// Dead letter topic const dlqConsumer = await client.subscribe({ topic: 'persistent://my-tenant/my-namespace/orders', subscription: 'order-processor', subscriptionType: 'Shared', deadLetterPolicy: { maxRedeliverCount: 3, deadLetterTopic: 'persistent://my-tenant/my-namespace/orders-dlq', }, });

Java

@Service public class OrderConsumer { @Autowired private PulsarClient client;

@PostConstruct
public void startConsumer() throws PulsarClientException {
    Consumer<byte[]> consumer = client.newConsumer()
        .topic("persistent://my-tenant/my-namespace/orders")
        .subscriptionName("order-processor")
        .subscriptionType(SubscriptionType.Shared)
        .ackTimeout(30, TimeUnit.SECONDS)
        .negativeAckRedeliveryDelay(1, TimeUnit.SECONDS)
        .deadLetterPolicy(DeadLetterPolicy.builder()
            .maxRedeliverCount(3)
            .deadLetterTopic("persistent://my-tenant/my-namespace/orders-dlq")
            .build())
        .subscribe();

    new Thread(() -> {
        while (true) {
            try {
                Message<byte[]> msg = consumer.receive();
                Order order = objectMapper.readValue(msg.getData(), Order.class);
                processOrder(order);
                consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        }
    }).start();
}

}

// Batch consumer Consumer<byte[]> batchConsumer = client.newConsumer() .topic("persistent://my-tenant/my-namespace/orders") .subscriptionName("batch-processor") .subscriptionType(SubscriptionType.Shared) .batchReceivePolicy(BatchReceivePolicy.builder() .maxNumMessages(100) .maxNumBytes(1024 * 1024) .timeout(100, TimeUnit.MILLISECONDS) .build()) .subscribe();

Messages<byte[]> messages = batchConsumer.batchReceive(); for (Message<byte[]> msg : messages) { processMessage(msg); } batchConsumer.acknowledge(messages);

Python

consumer = client.subscribe( 'persistent://my-tenant/my-namespace/orders', subscription_name='order-processor', consumer_type=pulsar.ConsumerType.Shared, dead_letter_policy=pulsar.DeadLetterPolicy( max_redeliver_count=3, dead_letter_topic='persistent://my-tenant/my-namespace/orders-dlq' ) )

while True: msg = consumer.receive() try: order = json.loads(msg.data().decode('utf-8')) process_order(order) consumer.acknowledge(msg) except Exception as e: consumer.negative_acknowledge(msg)

Topic Management

Partitioned topic

pulsar-admin topics create-partitioned-topic
persistent://my-tenant/my-namespace/orders
--partitions 12

Set retention

pulsar-admin namespaces set-retention my-tenant/my-namespace
--size 10G --time 7d

Set TTL

pulsar-admin namespaces set-message-ttl my-tenant/my-namespace
--messageTTL 86400

Compaction (for key-based topics)

pulsar-admin topics compact persistent://my-tenant/my-namespace/orders

When NOT to Use This Skill

Use alternative messaging solutions when:

• Simple request/reply patterns - RabbitMQ or NATS are simpler

• AWS-native architecture - SQS has better AWS integration

• Small-scale deployments - Operational complexity may not be justified

• JMS compliance required - Use ActiveMQ

• Extremely low latency (<1ms) - Use Redis or in-memory solutions

• Single datacenter only - Kafka may be simpler

• Limited operational expertise - Managed services may be better

Anti-Patterns

Anti-Pattern Why It's Bad Solution

Non-persistent topics in production Message loss on broker restart Use persistent topics

No backlog quotas Unbounded storage growth Set retention policies and quotas

Shared subscription for ordering Messages processed out of order Use Exclusive or Key_Shared

No dead letter topic Failed messages lost Configure DLT for all subscriptions

Large message batching Memory pressure, high latency Batch moderately (100-1000 messages)

Single bookie ensemble Data loss on bookie failure Use ensemble size >= 3

No namespace isolation Tenant resource contention Properly configure namespaces and quotas

Sync send everywhere Poor throughput Use async send with callbacks

No monitoring Invisible backlog and issues Monitor backlog, throughput, latency

Auto-ack without processing Message loss Acknowledge only after processing

Quick Troubleshooting

Issue Likely Cause Fix

Messages not delivered Subscription not created Create subscription before consuming

Backlog growing Slow consumers or consumer down Add consumers or optimize processing

High latency Batching delay or network Reduce batching delay, check network

Out of order messages Shared subscription Use Exclusive or Key_Shared

Connection timeout Broker overload or network Check broker health, network

Subscription not found Wrong topic or subscription name Verify names with pulsar-admin

Publish errors Topic not found or permissions Create topic, check ACLs

Geo-replication lag Network issues or high load Check replication backlog metrics

Bookie failures Disk full or hardware issues Monitor disk usage, replace bookie

Negative ack storm Processing failures Fix processing logic, add retry limits

Production Readiness

Security Configuration

broker.conf

authenticationEnabled=true authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken

brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken brokerClientAuthenticationParameters={"token":"xxx"}

TLS

tlsEnabled=true tlsCertificateFilePath=/path/to/broker.cert.pem tlsKeyFilePath=/path/to/broker.key-pk8.pem tlsTrustCertsFilePath=/path/to/ca.cert.pem

const client = new Pulsar.Client({ serviceUrl: 'pulsar+ssl://pulsar.example.com:6651', authentication: new Pulsar.AuthenticationToken({ token: 'xxx' }), tlsTrustCertsFilePath: '/path/to/ca.cert.pem', });

Geo-Replication

Enable replication

pulsar-admin namespaces set-clusters my-tenant/my-namespace
--clusters cluster-1,cluster-2,cluster-3

Check replication status

pulsar-admin topics stats persistent://my-tenant/my-namespace/orders

Monitoring Metrics

Metric Alert Threshold

Backlog size

100000 messages

Publish rate Anomaly detection

Subscription lag

10000

Message age

1 hour

Replication lag

1000

Checklist

• TLS enabled

• Authentication configured

• Authorization (namespace policies)

• Partitioned topics for scale

• Retention policies set

• Dead letter topics configured

• Geo-replication (if needed)

• Backlog quotas set

• Monitoring dashboards

• Backup procedures

Reference Documentation

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: pulsar for comprehensive documentation.

Available topics: basics , producers , consumers , production