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