Messaging Model

public interface Message<T> {
    // Core message components
    T getValue();              // Deserialized message payload
    byte[] getData();          // Raw payload bytes
    MessageId getMessageId();  // Unique message identifier
    
    // Metadata
    String getKey();                        // Optional message key
    byte[] getOrderingKey();                // Key for ordering in Key_Shared mode
    Map<String, String> getProperties();    // User-defined properties
    
    // Timestamps
    long getPublishTime();     // When producer sent the message
    long getEventTime();       // Application-defined event timestamp
    
    // Producer information
    String getProducerName();  // Name of the producing client
    String getTopicName();     // Topic where message was published
    
    // Advanced features
    long getSequenceId();              // Producer-assigned sequence
    int getRedeliveryCount();          // How many times redelivered
    byte[] getSchemaVersion();         // Schema version used
    Optional<EncryptionContext> getEncryptionCtx();  // Encryption info
}

// From TypedMessageBuilder interface
producer.newMessage()
    .value(myData)
    .key("user-123")
    .property("region", "us-west")
    .property("priority", "high")
    .eventTime(System.currentTimeMillis())
    .send();

// Producer sends without waiting for acknowledgment
producer.sendAsync(message);
// Don't wait for response - fire and forget

// Producer waits for broker acknowledgment
MessageId msgId = producer.send(message);

// Consumer acknowledges after processing
Message<String> msg = consumer.receive();
processMessage(msg);
consumer.acknowledge(msg);

// Using transactions for exactly-once semantics
Transaction txn = client.newTransaction()
    .withTransactionTimeout(1, TimeUnit.MINUTES)
    .build()
    .get();

try {
    // Produce and consume within transaction
    producer.newMessage(txn).value(data).send();
    consumer.acknowledgeAsync(msgId, txn).get();
    
    // Commit - messages visible atomically
    txn.commit().get();
} catch (Exception e) {
    // Abort - no messages are visible
    txn.abort().get();
}

Message<String> msg = consumer.receive();
try {
    processMessage(msg);
    consumer.acknowledge(msg);  // Acknowledge this message
} catch (Exception e) {
    consumer.negativeAcknowledge(msg);  // Redeliver this message
}

// Only works with Exclusive and Failover subscriptions
consumer.acknowledgeCumulative(msg);
// All messages up to 'msg' are now acknowledged

// From Consumer.java - negativeAcknowledge method
consumer.negativeAcknowledge(msg);
// Message will be redelivered after configured delay

// Configure dead letter policy
Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topic("my-topic")
    .subscriptionName("my-sub")
    .deadLetterPolicy(DeadLetterPolicy.builder()
        .maxRedeliverCount(3)
        .deadLetterTopic("my-topic-DLQ")
        .build())
    .subscribe();

// Use reconsumeLater for custom retry delays
consumer.reconsumeLater(msg, 5, TimeUnit.SECONDS);

# Set retention policy for a namespace
pulsar-admin namespaces set-retention my-tenant/my-namespace \
  --size 10G \
  --time 24h

Producer<byte[]> producer = client.newProducer()
    .topic("my-topic")
    .compressionType(CompressionType.LZ4)  // or ZLIB, ZSTD, SNAPPY
    .create();

Producer<String> producer = client.newProducer(Schema.STRING)
    .topic("my-topic")
    .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS)
    .batchingMaxMessages(1000)
    .batchingMaxBytes(1024 * 1024)  // 1 MB
    .create();

// Key-based routing (default for partitioned topics)
producer.newMessage()
    .key("user-123")
    .value(data)
    .send();
// All messages with same key go to same partition

// Custom routing
producer.newMessage()
    .orderingKey(customKey.getBytes())
    .value(data)
    .send();

// Enable at namespace level
pulsar-admin namespaces set-deduplication my-tenant/my-namespace --enable

// Producer sends with sequence ID
producer.newMessage()
    .value(data)
    .sequenceId(uniqueId)  // Must be monotonically increasing
    .send();