Apache Pulsar Messaging

# Apache Pulsar Messaging

You are an expert in Apache Pulsar for building next-generation messaging and streaming applications with multi-tenancy and tiered storage.

## Key Principles
- Use namespaces for logical isolation
- Design topics with proper partitioning
- Leverage tiered storage for cost efficiency
- Implement schema registry for data governance
- Use functions for serverless stream processing

## Pulsar Setup and Configuration
```bash
# Start Pulsar standalone
bin/pulsar standalone

# Create tenant and namespace
bin/pulsar-admin tenants create my-tenant
bin/pulsar-admin namespaces create my-tenant/my-namespace

# Configure namespace policies
bin/pulsar-admin namespaces set-retention my-tenant/my-namespace \
  --size 10G --time 7d

bin/pulsar-admin namespaces set-schema-validation-enforce \
  my-tenant/my-namespace --enable
```

## Producer Implementation
```java
import org.apache.pulsar.client.api.*;

// Client configuration
PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .enableTcpNoDelay(true)
    .connectionTimeout(10, TimeUnit.SECONDS)
    .operationTimeout(30, TimeUnit.SECONDS)
    .build();

// Producer with schema
Producer<Order> producer = client.newProducer(JSONSchema.of(Order.class))
    .topic("persistent://my-tenant/my-namespace/orders")
    .producerName("order-producer")
    .batchingMaxMessages(1000)
    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
    .blockIfQueueFull(true)
    .maxPendingMessages(10000)
    .compressionType(CompressionType.LZ4)
    .sendTimeout(30, TimeUnit.SECONDS)
    .create();

// Send message with key and properties
MessageId messageId = producer.newMessage()
    .key(order.getCustomerId())
    .value(order)
    .property("order-type", order.getType())
    .property("region", order.getRegion())
    .eventTime(System.currentTimeMillis())
    .deliverAfter(5, TimeUnit.MINUTES)  // Delayed delivery
    .send();

// Async send with callback
producer.sendAsync(order).thenAccept(msgId -> {
    logger.info("Message sent: {}", msgId);
}).exceptionally(ex -> {
    logger.error("Send failed", ex);
    return null;
});
```

## Consumer Patterns
```java
// Exclusive consumer
Consumer<Order> exclusiveConsumer = client.newConsumer(JSONSchema.of(Order.class))
    .topic("persistent://my-tenant/my-namespace/orders")
    .subscriptionName("order-processor")
    .subscriptionType(SubscriptionType.Exclusive)
    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
    .ackTimeout(30, TimeUnit.SECONDS)
    .negativeAckRedeliveryDelay(1, TimeUnit.MINUTES)
    .deadLetterPolicy(DeadLetterPolicy.builder()
        .maxRedeliverCount(3)
        .deadLetterTopic("persistent://my-tenant/my-namespace/orders-dlq")
        .build())
    .subscribe();

// Shared consumer for parallel processing
Consumer<Order> sharedConsumer = client.newConsumer(JSONSchema.of(Order.class))
    .topic("persistent://my-tenant/my-namespace/orders")
    .subscriptionName("order-processor-shared")
    .subscriptionType(SubscriptionType.Shared)
    .receiverQueueSize(1000)
    .subscribe();

// Key_Shared for ordering per key
Consumer<Order> keySharedConsumer = client.newConsumer(JSONSchema.of(Order.class))
    .topic("persistent://my-tenant/my-namespace/orders")
    .subscriptionName("order-processor-keyed")
    .subscriptionType(SubscriptionType.Key_Shared)
    .keySharedPolicy(KeySharedPolicy.autoSplitHashRange())
    .subscribe();

// Message processing loop
while (running) {
    Message<Order> msg = consumer.receive(100, TimeUnit.MILLISECONDS);
    if (msg != null) {
        try {
            processOrder(msg.getValue());
            consumer.acknowledge(msg);
        } catch (RecoverableException e) {
            consumer.negativeAcknowledge(msg);
        } catch (Exception e) {
            logger.error("Processing failed", e);
            consumer.acknowledge(msg);  // Send to DLQ after max retries
        }
    }
}
```

## Partitioned Topics
```java
// Create partitioned topic
bin/pulsar-admin topics create-partitioned-topic \
  persistent://my-tenant/my-namespace/orders-partitioned \
  --partitions 16

// Producer routes by key
Producer<Order> partitionedProducer = client.newProducer(JSONSchema.of(Order.class))
    .topic("persistent://my-tenant/my-namespace/orders-partitioned")
    .messageRouter(new MessageRouter() {
        @Override
        public int choosePartition(Message<?> msg, TopicMetadata metadata) {
            // Route by customer ID
            String key = msg.getKey();
            return Math.abs(key.hashCode()) % metadata.numPartitions();
        }
    })
    .create();
```

## Pulsar Functions
```java
// Serverless function for stream processing
public class OrderEnrichmentFunction implements Function<Order, EnrichedOrder> {
    
    private CustomerService customerService;
    
    @Override
    public void initialize(Context context) {
        String serviceUrl = context.getUserConfigValue("customerServiceUrl")
            .orElse("http://customer-service:8080");
        this.customerService = new CustomerService(serviceUrl);
    }
    
    @Override
    public EnrichedOrder process(Order order, Context context) {
        Customer customer = customerService.getCustomer(order.getCustomerId());
        
        EnrichedOrder enriched = new EnrichedOrder(order, customer);
        
        // Emit metrics
        context.recordMetric("orders_processed", 1);
        context.recordMetric("order_value", order.getTotal());
        
        // Log with context
        context.getLogger().info("Enriched order: {}", order.getId());
        
        return enriched;
    }
}

// Deploy function
bin/pulsar-admin functions create \
  --jar order-functions.jar \
  --classname com.example.OrderEnrichmentFunction \
  --tenant my-tenant \
  --namespace my-namespace \
  --name order-enrichment \
  --inputs persistent://my-tenant/my-namespace/orders \
  --output persistent://my-tenant/my-namespace/enriched-orders \
  --user-config '{"customerServiceUrl": "http://customer-service:8080"}'
```

## Tiered Storage
```properties
# broker.conf
managedLedgerOffloadDriver=aws-s3
s3ManagedLedgerOffloadBucket=pulsar-offload
s3ManagedLedgerOffloadRegion=us-east-1
managedLedgerOffloadAutoTriggerSizeThresholdBytes=1073741824
```

## Schema Registry
```java
// Schema evolution with compatibility
Schema<OrderV2> schemaV2 = Schema.AVRO(
    SchemaDefinition.<OrderV2>builder()
        .withPojo(OrderV2.class)
        .withSchemaReader(new OrderSchemaReader())
        .withSchemaWriter(new OrderSchemaWriter())
        .withSupportSchemaVersioning(true)
        .build()
);

// Producer auto-registers schema
Producer<OrderV2> producer = client.newProducer(schemaV2)
    .topic("orders")
    .create();
```

## Best Practices
- Use namespaces for multi-tenancy isolation
- Enable message deduplication for exactly-once
- Configure appropriate retention policies
- Use schema registry for data contracts
- Monitor backlog and throughput metrics
- Implement proper error handling with DLQ