pulsar

What this skill does

# Apache Pulsar Core Knowledge

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

## Quick Start (Docker)

```yaml
# 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:
```

```bash
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)
```typescript
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)
```java
@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)
```python
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)
```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
```typescript
// 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
```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-n