Internals: How Pulsar 3.2 Implements Tiered Storage for Event Streaming

import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.tieredstorage.TieredStorageConfiguration;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.ReadHandle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Core offload trigger logic for Pulsar 3.2 Tiered Storage.
 * Evaluates namespace policies and ledger metadata to decide if a segment should be offloaded.
 * This class is instantiated per ManagedLedger and runs in the broker's IO thread.
 */
public class OffloadTrigger {
    private static final Logger log = LoggerFactory.getLogger(OffloadTrigger.class);
    private final NamespaceName namespace;
    private final ServiceConfiguration brokerConfig;
    private final TieredStorageConfiguration tsConfig;
    private final ScheduledExecutorService executor;
    private volatile OffloadPolicies activePolicies;

    public OffloadTrigger(NamespaceName namespace, ServiceConfiguration brokerConfig,
                          TieredStorageConfiguration tsConfig, ScheduledExecutorService executor) {
        this.namespace = namespace;
        this.brokerConfig = brokerConfig;
        this.tsConfig = tsConfig;
        this.executor = executor;
        // Load initial offload policies from namespace admin API
        this.activePolicies = brokerConfig.getOffloadPolicies();
        log.info("Initialized offload trigger for namespace {} with policies: {}", namespace, activePolicies);
    }

    /**
     * Evaluates whether a given ledger segment should be offloaded to remote storage.
     * @param ledgerHandle Read handle to the BookKeeper ledger containing the segment
     * @param segmentStartEntryId First entry ID of the segment
     * @param segmentEndEntryId Last entry ID of the segment
     * @param segmentAgeMs Age of the segment in milliseconds since creation
     * @return CompletableFuture indicating if offload should proceed
     */
    public CompletableFuture shouldOffload(ReadHandle ledgerHandle, long segmentStartEntryId,
                                                    long segmentEndEntryId, long segmentAgeMs) {
        CompletableFuture resultFuture = new CompletableFuture<>();

        // Run policy evaluation off the IO thread to avoid blocking ledger reads
        executor.submit(() -> {
            try {
                // 1. Check if tiered storage is enabled globally and for this namespace
                if (!tsConfig.isTieredStorageEnabled() || !activePolicies.isOffloadEnabled()) {
                    resultFuture.complete(false);
                    return;
                }

                // 2. Check minimum segment age threshold (default 1 hour in Pulsar 3.2)
                long minAgeMs = activePolicies.getOffloadMinAgeMs();
                if (segmentAgeMs < minAgeMs) {
                    log.debug("Segment {}-{} in namespace {} too young to offload: {}ms < {}ms",
                            segmentStartEntryId, segmentEndEntryId, namespace, segmentAgeMs, minAgeMs);
                    resultFuture.complete(false);
                    return;
                }

                // 3. Check segment size threshold (default 1GB in Pulsar 3.2)
                long segmentSizeBytes = (segmentEndEntryId - segmentStartEntryId + 1) * brokerConfig.getManagedLedgerEntrySizeThreshold();
                long minSizeBytes = activePolicies.getOffloadMinSizeBytes();
                if (segmentSizeBytes < minSizeBytes) {
                    log.debug("Segment {}-{} in namespace {} too small to offload: {} bytes < {} bytes",
                            segmentStartEntryId, segmentEndEntryId, namespace, segmentSizeBytes, minSizeBytes);
                    resultFuture.complete(false);
                    return;
                }

                // 4. Check if ledger is closed (only closed ledgers can be offloaded in Pulsar 3.2)
                if (!ledgerHandle.isClosed()) {
                    log.debug("Ledger {} for segment {}-{} is still open, skipping offload",
                            ledgerHandle.getId(), segmentStartEntryId, segmentEndEntryId);
                    resultFuture.complete(false);
                    return;
                }

                // 5. Check remote storage connectivity (lightweight ping, not full write)
                tsConfig.getStorageProvider().ping().thenAccept(pingSuccess -> {
                    if (pingSuccess) {
                        log.info("Triggering offload for segment {}-{} in namespace {}, size: {} bytes, age: {}ms",
                                segmentStartEntryId, segmentEndEntryId, namespace, segmentSizeBytes, segmentAgeMs);
                        resultFuture.complete(true);
                    } else {
                        log.warn("Remote storage ping failed for namespace {}, skipping offload", namespace);
                        resultFuture.complete(false);
                    }
                }).exceptionally(ex -> {
                    log.error("Error pinging remote storage for namespace {}", namespace, ex);
                    resultFuture.complete(false);
                    return null;
                });

            } catch (Exception e) {
                log.error("Unexpected error evaluating offload for segment {}-{} in namespace {}",
                        segmentStartEntryId, segmentEndEntryId, namespace, e);
                resultFuture.completeExceptionally(e);
            }
        });

        return resultFuture;
    }

    /**
     * Updates active offload policies when namespace config changes.
     * @param newPolicies Updated offload policies from admin API
     */
    public void updatePolicies(OffloadPolicies newPolicies) {
        this.activePolicies = newPolicies;
        log.info("Updated offload policies for namespace {}: {}", namespace, newPolicies);
    }
}

import org.apache.pulsar.broker.tieredstorage.providers.TieredStorageProvider;
import org.apache.pulsar.broker.tieredstorage.SegmentInfo;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.S3Exception;
import org.apache.bookkeeper.client.ReadHandle;
import org.apache.bookkeeper.util.IOUtils;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;

/**
 * S3-compatible implementation of TieredStorageProvider for Pulsar 3.2.
 * Handles writing offloaded segments to S3/GCS/Azure Blob (all S3-compatible APIs).
 */
public class S3StorageProvider implements TieredStorageProvider {
    private final S3Client s3Client;
    private final String bucketName;
    private final String region;
    private final int maxRetries;
    private final int retryDelayMs;

    public S3StorageProvider(S3Client s3Client, String bucketName, String region,
                             int maxRetries, int retryDelayMs) {
        this.s3Client = s3Client;
        this.bucketName = bucketName;
        this.region = region;
        this.maxRetries = maxRetries;
        this.retryDelayMs = retryDelayMs;
    }

    @Override
    public CompletableFuture writeSegment(SegmentInfo segmentInfo, ReadHandle ledgerHandle) {
        CompletableFuture writeFuture = new CompletableFuture<>();

        // Run write in dedicated storage thread pool to avoid blocking broker IO
        CompletableFuture.runAsync(() -> {
            int retryCount = 0;
            boolean success = false;

            while (retryCount <= maxRetries && !success) {
                try {
                    // 1. Read segment data from BookKeeper ledger
                    ByteArrayOutputStream segmentDataStream = new ByteArrayOutputStream();
                    long startEntryId = segmentInfo.getStartEntryId();
                    long endEntryId = segmentInfo.getEndEntryId();
                    long entryCount = endEntryId - startEntryId + 1;

                    for (long entryId = startEntryId; entryId <= endEntryId; entryId++) {
                        // Read entry from BookKeeper with 3 retries per entry
                        ByteBuffer entryBuffer = readEntryWithRetries(ledgerHandle, entryId, 3);
                        segmentDataStream.write(entryBuffer.array(), entryBuffer.arrayOffset(), entryBuffer.remaining());
                    }

                    byte[] segmentData = segmentDataStream.toByteArray();
                    segmentDataStream.close();

                    // 2. Construct S3 object key: namespace/topic/ledgerId/segmentStart-segmentEnd
                    String objectKey = buildObjectKey(segmentInfo);

                    // 3. Create S3 PutObject request with metadata (entry count, checksum, ledger ID)
                    PutObjectRequest putRequest = PutObjectRequest.builder()
                            .bucket(bucketName)
                            .key(objectKey)
                            .metadata("pulsar-namespace", segmentInfo.getNamespace().toString())
                            .metadata("pulsar-topic", segmentInfo.getTopic().toString())
                            .metadata("pulsar-ledger-id", String.valueOf(segmentInfo.getLedgerId()))
                            .metadata("pulsar-entry-count", String.valueOf(entryCount))
                            .metadata("pulsar-segment-checksum", String.valueOf(segmentInfo.getChecksum()))
                            .build();

                    // 4. Write segment to S3 with request body
                    s3Client.putObject(putRequest, RequestBody.fromBytes(segmentData));

                    log.info("Successfully wrote segment {} to S3 bucket {} key {}", segmentInfo, bucketName, objectKey);
                    success = true;
                    writeFuture.complete(null);

                } catch (S3Exception e) {
                    retryCount++;
                    log.warn("S3 write failed for segment {} (attempt {}/{}): {}",
                            segmentInfo, retryCount, maxRetries, e.getMessage());
                    if (retryCount > maxRetries) {
                        log.error("Exhausted retries for S3 write of segment {}", segmentInfo, e);
                        writeFuture.completeExceptionally(new IOException("Failed to write segment to S3 after " + maxRetries + " retries", e));
                    } else {
                        try {
                            Thread.sleep(retryDelayMs * retryCount); // Exponential backoff would be better, but simplified here
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                            writeFuture.completeExceptionally(ie);
                        }
                    }
                } catch (IOException e) {
                    log.error("IO error writing segment {} to S3", segmentInfo, e);
                    writeFuture.completeExceptionally(e);
                } catch (Exception e) {
                    log.error("Unexpected error writing segment {} to S3", segmentInfo, e);
                    writeFuture.completeExceptionally(e);
                }
            }
        });

        return writeFuture;
    }

    private ByteBuffer readEntryWithRetries(ReadHandle ledgerHandle, long entryId, int retries) throws IOException {
        int attempt = 0;
        while (attempt <= retries) {
            try {
                return ledgerHandle.readEntry(entryId);
            } catch (Exception e) {
                attempt++;
                if (attempt > retries) throw new IOException("Failed to read entry " + entryId + " after " + retries + " retries", e);
                try { Thread.sleep(100 * attempt); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw new IOException(ie); }
            }
        }
        throw new IOException("Failed to read entry " + entryId);
    }

    private String buildObjectKey(SegmentInfo segmentInfo) {
        return String.format("%s/%s/%d/%d-%d",
                segmentInfo.getNamespace().getTenant(),
                segmentInfo.getNamespace().getCluster(),
                segmentInfo.getLedgerId(),
                segmentInfo.getStartEntryId(),
                segmentInfo.getEndEntryId());
    }

    @Override
    public CompletableFuture ping() {
        // Simplified ping: list objects with max 1 key to check connectivity
        return CompletableFuture.supplyAsync(() -> {
            try {
                s3Client.listObjectsV2(b -> b.bucket(bucketName).maxKeys(1));
                return true;
            } catch (S3Exception e) {
                log.warn("S3 ping failed for bucket {}", bucketName, e);
                return false;
            }
        });
    }
}

import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.bookkeeper.client.ReadHandle;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.S3Exception;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

/**
 * InputStream implementation that reads offloaded segments from remote storage,
 * with fallback to BookKeeper if remote storage is unavailable.
 * Used by Pulsar consumers to transparently read both hot and offloaded data.
 */
public class TieredStorageInputStream extends InputStream {
    private static final int BUFFER_SIZE = 1024 * 1024; // 1MB buffer for remote reads
    private final SegmentInfo segmentInfo;
    private final S3Client s3Client;
    private final String bucketName;
    private final ReadHandle ledgerFallbackHandle;
    private final ExecutorService readerExecutor;
    private volatile InputStream remoteInputStream;
    private volatile boolean useFallback = false;
    private long bytesRead = 0;
    private final long totalSegmentSize;

    public TieredStorageInputStream(SegmentInfo segmentInfo, S3Client s3Client, String bucketName,
                                    ReadHandle ledgerFallbackHandle, ExecutorService readerExecutor) {
        this.segmentInfo = segmentInfo;
        this.s3Client = s3Client;
        this.bucketName = bucketName;
        this.ledgerFallbackHandle = ledgerFallbackHandle;
        this.readerExecutor = readerExecutor;
        this.totalSegmentSize = segmentInfo.getSegmentSizeBytes();
    }

    @Override
    public int read() throws IOException {
        // Single byte read: delegate to buffer read for efficiency
        byte[] buffer = new byte[1];
        int bytesRead = read(buffer, 0, 1);
        if (bytesRead == -1) return -1;
        return buffer[0] & 0xFF;
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException {
        if (useFallback) {
            // Fallback to BookKeeper: read entry by entry (simplified for example)
            return readFromBookKeeperFallback(b, off, len);
        }

        try {
            // Lazy initialize remote input stream
            if (remoteInputStream == null) {
                synchronized (this) {
                    if (remoteInputStream == null) {
                        String objectKey = buildObjectKey(segmentInfo);
                        GetObjectRequest getRequest = GetObjectRequest.builder()
                                .bucket(bucketName)
                                .key(objectKey)
                                .build();
                        remoteInputStream = s3Client.getObject(getRequest);
                        log.info("Opened remote input stream for segment {} key {}", segmentInfo, objectKey);
                    }
                }
            }

            int bytesReadNow = remoteInputStream.read(b, off, len);
            if (bytesReadNow == -1) {
                // End of remote stream: close and return -1
                close();
                return -1;
            }
            bytesRead += bytesReadNow;
            return bytesReadNow;

        } catch (S3Exception e) {
            log.warn("Failed to read from remote storage for segment {}, falling back to BookKeeper: {}",
                    segmentInfo, e.getMessage());
            useFallback = true;
            closeQuietly(remoteInputStream);
            return readFromBookKeeperFallback(b, off, len);
        } catch (IOException e) {
            log.error("IO error reading from remote storage for segment {}", segmentInfo, e);
            throw e;
        }
    }

    private int readFromBookKeeperFallback(byte[] b, int off, int len) throws IOException {
        // Simplified fallback: read entries sequentially from BookKeeper
        // In real Pulsar 3.2, this uses the ManagedLedger's entry cache and read path
        long currentEntryId = segmentInfo.getStartEntryId() + (bytesRead / segmentInfo.getEntrySizeAvg());
        if (currentEntryId > segmentInfo.getEndEntryId()) {
            return -1;
        }
        try {
            ByteBuffer entryBuffer = ledgerFallbackHandle.readEntry(currentEntryId);
            int bytesToCopy = Math.min(len, entryBuffer.remaining());
            entryBuffer.get(b, off, bytesToCopy);
            bytesRead += bytesToCopy;
            return bytesToCopy;
        } catch (Exception e) {
            throw new IOException("Failed to read entry " + currentEntryId + " from BookKeeper fallback", e);
        }
    }

    private String buildObjectKey(SegmentInfo segmentInfo) {
        return String.format("%s/%s/%d/%d-%d",
                segmentInfo.getNamespace().getTenant(),
                segmentInfo.getNamespace().getCluster(),
                segmentInfo.getLedgerId(),
                segmentInfo.getStartEntryId(),
                segmentInfo.getEndEntryId());
    }

    @Override
    public void close() throws IOException {
        closeQuietly(remoteInputStream);
        // Don't close ledger fallback handle: it's managed by the ManagedLedger
        remoteInputStream = null;
        log.info("Closed tiered storage input stream for segment {}, total bytes read: {}", segmentInfo, bytesRead);
    }

    private void closeQuietly(InputStream is) {
        if (is != null) {
            try { is.close(); } catch (IOException e) { log.warn("Error closing input stream", e); }
        }
    }

    @Override
    public long skip(long n) throws IOException {
        // Simplified skip: read and discard bytes
        byte[] skipBuffer = new byte[BUFFER_SIZE];
        long remaining = n;
        while (remaining > 0) {
            int bytesSkipped = read(skipBuffer, 0, (int) Math.min(remaining, BUFFER_SIZE));
            if (bytesSkipped == -1) break;
            remaining -= bytesSkipped;
        }
        return n - remaining;
    }

    @Override
    public boolean markSupported() {
        return false; // Mark/reset not supported for remote or fallback streams
    }
}

curl -s http://broker:8080/metrics | grep pulsar_tiered_storage_offload_worker_active_threads

tieredStorageProvider=s3
bucketName=pulsar-offload-intelligent-tiering
region=us-east-1
endpoint=https://s3.us-east-1.amazonaws.com

pulsar-admin namespaces set-offload-encryption-key \
  --key-id arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab \
  my-tenant/my-namespace