Pulsar

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

Architecture

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

Subscription Types

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

Quick Troubleshooting

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

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