SprintReady read a real specification in minutes, and on day one caught problems the real team only hit years later.
Apache Kafka is open-source software used by thousands of companies. Its engineers designed a major feature, Tiered Storage, in a public document and recorded every piece of work on public Jira. We gave SprintReady the document as it stood on 12 February 2020, the day before the team opened its first ticket, then followed what the real team actually built, fixed and discovered over the next three and a half years.
Source: the Apache Kafka project's public design document (KIP-405) and Jira tickets, used under the Apache License 2.0, with names replaced by labels. SprintReady is not affiliated with or endorsed by the Apache Software Foundation. Full legal notice.
Caught on day one. Found by the team months or years later.
A problem in the specification is cheapest to fix before anyone writes code. These are behaviours SprintReady wrote into testable acceptance criteria, or raised as questions, on 12 February 2020, and the tickets the Kafka team later had to open when the same problems turned up in the work.
The offset rules are validated before and after every remote segment deletion; any operation that would break them is rejected or deferred.
Remote segments were being deleted before the log start offset was updated, so a consumer reading from the beginning got UNKNOWN_SERVER_ERROR. Marked a Blocker for the 3.6.0 release, weeks before it shipped.
Duplicate segments resulting from a leader-to-follower transition are handled during remote retention clean-up.
After a leader change the remote log manager could upload duplicate segments, which could affect followers and consumers. Fixed for the 3.6.0 release.
Question: the local file format and location for storing the metadata cache's committed offsets aren't specified.
The file-based cache was settled 15 months later, missed the 3.0 feature freeze and was pushed to the next release.
Fixing a problem late means a new ticket, a code change, a review and a re-test, under release pressure. With SprintReady's tickets in the backlog from day one, each of these would have been a test case to pass, not a problem to discover.
What it saves
The assumptions are the same ones in the cost calculator on our home page: the cost-saving calculator on sprintready.dev (25 minutes a ticket; £450 a day senior BA; 2.7 credits a ticket at £399 per 500 credits). Change them there to your own numbers.
Asked on day one. Settled by the team over a year later.
SprintReady doesn't fill a gap with a guess: it writes the ticket from what the document says and raises a precise question for what it doesn't. These are gaps it raised on 12 February 2020 that the Kafka team later had to settle in tickets of its own.
Default values for six tiered-storage settings (thread pools, task interval, reader queue, remote retention time and size)
Names and replication factor for the internal metadata topic
Schema of the internal topic's messages
Local file format for storing the metadata cache's committed offsets
See all 11 questions, and the ones we didn't count
An AI check proposed 20 links between SprintReady's questions and later tickets. We read each one against the real ticket and kept only those where the later ticket settles the same specific point.
Kept (11):
- SR-6.1 CLARIFICATION NEEDED: Default value for remote.log.manager.thread.pool.size not specified → KAFKA-12816
- SR-7.1 CLARIFICATION NEEDED: Default value for remote.log.manager.task.interval.ms not specified → KAFKA-12816
- SR-8.1 CLARIFICATION NEEDED: Default value for remote.log.reader.threads not specified → KAFKA-12816
- SR-9.1 CLARIFICATION NEEDED: Default value for remote.log.reader.max.pending.tasks not specified → KAFKA-12816
- SR-10.1 CLARIFICATION NEEDED: Default value for remote.log.retention.minutes not specified → KAFKA-12816
- SR-11.1 CLARIFICATION NEEDED: Default value for remote.log.retention.bytes not specified → KAFKA-12816
- SR-17.1 CLARIFICATION NEEDED: Exact name and schema of the __remote_segments_to_be_deleted topic not fully specified → KAFKA-12429
- SR-29.1 CLARIFICATION NEEDED: Exact configuration property names for topic name, partition count, and replication factor not specified → KAFKA-12816
- SR-29.2 CLARIFICATION NEEDED: Default replication factor not specified → KAFKA-12816
- SR-31.2 CLARIFICATION NEEDED: Local file format and location for storing committed offsets not specified → KAFKA-12802
- SR-35.2 CLARIFICATION NEEDED: Complete topic schema (key, value format) not specified → KAFKA-12429
Not counted (9):
- SR-6.2 CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified (excluded: not the same specific point, or not confirmable from the ticket)
- SR-7.2 CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified (excluded: not the same specific point, or not confirmable from the ticket)
- SR-8.2 CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified (excluded: not the same specific point, or not confirmable from the ticket)
- SR-9.2 CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified (excluded: not the same specific point, or not confirmable from the ticket)
- SR-10.2 CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified (excluded: not the same specific point, or not confirmable from the ticket)
- SR-11.2 CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified (excluded: not the same specific point, or not confirmable from the ticket)
- SR-13.1 CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]") (excluded: caused by our conversion (code blocks removed))
- SR-14.1 CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]") (excluded: caused by our conversion (code blocks removed))
- SR-34.1 CLARIFICATION NEEDED: Configurable maximum cache size (number of files or bytes) not specified (excluded: not the same specific point, or not confirmable from the ticket)
Raised before the work starts, a question like these costs a conversation. Found during the work, it costs a ticket, a change and a re-test.
How complete was it?
A fair test compares SprintReady with the work the document actually described. Of the 65 tickets the team opened after 12 February 2020:
- 13 planned by SprintReady: the core design, from the plug-in interfaces to copying, reading and deleting remote data, follower replication and configuration.
- 1 described but not planned: a simple test implementation the document mentions in one sentence. Our one genuine miss.
- 2 outside the project: the S3 and HDFS plug-ins, which the document says will live in external repositories.
- 1 closed as Won't Fix: a report that turned out not to be a problem.
- 48 added after February 2020: new scope, test frameworks, metrics, support for Kafka's later KRaft mode, and fixes found in 2023. No tool can write tickets for work a specification doesn't contain yet; when the document changes, SprintReady updates the backlog from the new version.
Each later ticket checked against the 12 Feb 2020 document by a strict AI check that had to quote the document (quotes verified in code), then read by hand.
Compare every ticket
Click any ticket to see the two side by side.
2020-02-13 KAFKA-9548 SPI - RemoteStorageManager and RemoteLogMetadataManager interfaces and related classes. Planned by SprintReady
Sub-task.
Define the RemoteStorageManager interface as specified in the design document. This interface provides the lifecycle operations for remote log segments and indexes, including copying log segments to remote storage, fetching log segment data, and deleting remote log segments. The interface must be the integration point for pluggable remote storage backend implementations.
Acceptance criteria (7)
- A Java interface named RemoteStorageManager is created with all methods specified in the design document
- The interface includes a copyLogSegment method that accepts RemoteLogSegmentId and LogSegmentData and returns success/failure indication
- The interface includes a fetchLogSegmentData method that accepts RemoteLogSegmentMetadata, startPosition (Long), and optional endPosition (Optional<Long>) and returns a Records instance
- The interface includes a deleteLogSegment method that accepts RemoteLogSegmentId
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > Remote Storage Manager", statement "RemoteStorageManager is an interface to provide the lifecycle of remote log segments and indexes." and "[code omitted]"
Define the RemoteLogMetadataManager interface as specified in the design document. This interface provides the lifecycle operations for metadata about remote log segments with strongly consistent semantics. The interface must support storing, fetching, and managing remote log segment metadata, and must allow for pluggable custom implementations while providing a default implementation backed by an internal Kafka topi…
Acceptance criteria (8)
- A Java interface named RemoteLogMetadataManager is created with all methods specified in the design document
- The interface includes a putRemoteLogSegmentData method that accepts RemoteLogSegmentId and RemoteLogSegmentMetadata
- The interface includes methods to fetch remote log segment metadata for a given offset
- The interface includes a method to register a topic partition for metadata cache building
- The interface includes an earliestLogOffset method that accepts TopicPartition and returns the earliest offset tracked in remote storage
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > RemoteLogMetadataManager", statement "RemoteLogMetadataManager is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics." and "[code omitted]"
Why it matches: Covers defining the RemoteStorageManager and RemoteLogMetadataManager interfaces as specified in the design. (checked by hand)
2020-02-13 KAFKA-9549 Local storage implementations for RSM which can be used in tests Described, not planned
Sub-task. The goal of this task is to implement a straightforward file-system based implementation of the {{RemoteStorageManager}} defined as part of the SPI for Tiered Storage. It is intended to be used in single-host integration tests where the remote storage is or can be exercised.
The document says: "We will provide a simple implementation of RSM to get a better understanding of the APIs." SprintReady didn't write a ticket for it: our one genuine miss.
2020-02-13 KAFKA-9550 RemoteLogManager - copying eligible log segments to remote storage implementation Planned by SprintReady
Sub-task. Implementation of RLM as mentioned in the HLD section of KIP-405, this JIRA covers copying segments to remote storage. [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP-405:KafkaTieredStorage-High-leveldesign]
Implement the RLM Leader Task that runs for each leader topic partition. This task is responsible for identifying inactive log segment files that meet the criteria for remote upload and copying them to remote storage. The task must process segments one by one from earliest to latest, generate a unique RemoteLogSegmentId for each segment, store metadata via RemoteLogMetadataManager, and invoke copyLogSegment on Remote…
Acceptance criteria (11)
- For each topic-partition where the broker is the leader, an RLM Leader Task is created and scheduled
- The task identifies inactive log segment files (not the active segment)
- The task filters segments where the offset range is not already covered by segments in remote storage
- The task filters segments where the last offset is less than the last-stable-offset of the partition
- Eligible segments are processed in order from earliest to latest
- For each segment, a universally unique RemoteLogSegmentId is generated
- The task calls RemoteLogMetadataManager.putRemoteLogSegmentData with the RemoteLogSegmentId and RemoteLogSegmentMetadata
- The task calls RemoteStorageManager.copyLogSegment with the RemoteLogSegmentId and LogSegmentData
- If copyLogSegment is successful, the metadata is updated with success status
- If copyLogSegment fails, the metadata entry is removed
- The task handles failures gracefully and schedules retry according to the configured retry interval
Traced to: KIP-405 Section "Proposed Changes > High-level design" and "RLM/RSM tasks and thread pools > 1. Remote Log Manager (RLM) Thread Pool > Copy log segments to remote storage (leader)", full paragraph starting with "Copy the log segment files that are inactive..."
Why it matches: Covers the RLM Leader Task for copying eligible log segments to remote storage. (checked by hand)
2020-02-14 KAFKA-9554 Define the SPI for Tiered Storage framework Planned by SprintReady
Sub-task. The goal of this task is to define the SPI (service provider interfaces) which will be used by vendors to implement plug-ins to communicate with specific storage system. Done means: * Package with interfaces and key objects available and published for review.
Define the RemoteStorageManager interface as specified in the design document. This interface provides the lifecycle operations for remote log segments and indexes, including copying log segments to remote storage, fetching log segment data, and deleting remote log segments. The interface must be the integration point for pluggable remote storage backend implementations.
Acceptance criteria (7)
- A Java interface named RemoteStorageManager is created with all methods specified in the design document
- The interface includes a copyLogSegment method that accepts RemoteLogSegmentId and LogSegmentData and returns success/failure indication
- The interface includes a fetchLogSegmentData method that accepts RemoteLogSegmentMetadata, startPosition (Long), and optional endPosition (Optional<Long>) and returns a Records instance
- The interface includes a deleteLogSegment method that accepts RemoteLogSegmentId
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > Remote Storage Manager", statement "RemoteStorageManager is an interface to provide the lifecycle of remote log segments and indexes." and "[code omitted]"
Define the RemoteLogMetadataManager interface as specified in the design document. This interface provides the lifecycle operations for metadata about remote log segments with strongly consistent semantics. The interface must support storing, fetching, and managing remote log segment metadata, and must allow for pluggable custom implementations while providing a default implementation backed by an internal Kafka topi…
Acceptance criteria (8)
- A Java interface named RemoteLogMetadataManager is created with all methods specified in the design document
- The interface includes a putRemoteLogSegmentData method that accepts RemoteLogSegmentId and RemoteLogSegmentMetadata
- The interface includes methods to fetch remote log segment metadata for a given offset
- The interface includes a method to register a topic partition for metadata cache building
- The interface includes an earliestLogOffset method that accepts TopicPartition and returns the earliest offset tracked in remote storage
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > RemoteLogMetadataManager", statement "RemoteLogMetadataManager is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics." and "[code omitted]"
Why it matches: Defining the SPI interfaces matches the RemoteStorageManager and RemoteLogMetadataManager interface specifications. (checked by hand)
2020-02-14 KAFKA-9555 Topic-based implementation for the RemoteLogMetadataManager Planned by SprintReady
Sub-task. The purpose of this task is to implement a {{RemoteLogMetadataManager}} based on an internal topic in Kafka. More details ar mentioned in the [KIP|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-RemoteLogMetadataManagerimplementedwithaninternaltopic]. Done means: - Pull Request available for review and unit-tests. System and integration tests are out…
Add an optional system-wide configuration property `remote.log.metadata.manager.class.name` that specifies the fully qualified class name of a custom RemoteLogMetadataManager implementation. If this property is not configured, the broker must use a default inbuilt metadata manager backed by an internal Kafka topic. The configuration must be validated at startup if provided.
Acceptance criteria (7)
- A new configuration property `remote.log.metadata.manager.class.name` is added to the broker configuration schema
- The property is optional
- The property accepts a fully qualified Java class name as a string value
- If the property is not set, the broker initializes the default RemoteLogMetadataManager implementation backed by an internal topic
- If the property is set, the specified class is validated at startup to ensure it exists on the classpath and implements the RemoteLogMetadataManager interface
- If the class does not exist or does not implement RemoteLogMetadataManager, broker startup fails with a descriptive error
- The default internal-topic-based metadata manager is documented as the fallback behavior
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.metadata.manager.class.name(optional) - This is an optional property. If this is not configured, Kafka uses an inbuilt metadata manager backed by an internal topic."
Define the RemoteLogMetadataManager interface as specified in the design document. This interface provides the lifecycle operations for metadata about remote log segments with strongly consistent semantics. The interface must support storing, fetching, and managing remote log segment metadata, and must allow for pluggable custom implementations while providing a default implementation backed by an internal Kafka topi…
Acceptance criteria (8)
- A Java interface named RemoteLogMetadataManager is created with all methods specified in the design document
- The interface includes a putRemoteLogSegmentData method that accepts RemoteLogSegmentId and RemoteLogSegmentMetadata
- The interface includes methods to fetch remote log segment metadata for a given offset
- The interface includes a method to register a topic partition for metadata cache building
- The interface includes an earliestLogOffset method that accepts TopicPartition and returns the earliest offset tracked in remote storage
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > RemoteLogMetadataManager", statement "RemoteLogMetadataManager is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics." and "[code omitted]"
Why it matches: Topic-based RemoteLogMetadataManager implementation is the default implementation described in SR-3 and SR-14. (checked by hand)
2020-02-17 KAFKA-9564 Integration Test framework for Tiered Storage Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2020-02-18 KAFKA-9565 Implementation of Tiered Storage SPI to integrate with S3 Outside the project (per the document)
Sub-task.
The document says S3 and HDFS implementations "are planned to be hosted in external repos", outside the project, so SprintReady rightly didn't plan them.
2020-02-18 KAFKA-9569 RemoteStorageManager implementation for HDFS storage. Outside the project (per the document)
Sub-task. This is about implementing `RemoteStorageManager` for HDFS to verify the proposed SPIs are sufficient. It looks like the existing RSM interface should be sufficient. If needed, we will discuss any required changes.
The document says S3 and HDFS implementations "are planned to be hosted in external repos", outside the project, so SprintReady rightly didn't plan them.
2020-02-20 KAFKA-9579 Remote consumer fetch implementation by adding respective purgatory Planned by SprintReady
Sub-task.
Implement the remote storage fetcher thread pool that processes consumer fetch requests requiring data from remote storage. When a fetch request targets an offset in remote storage, the request is added to a RemoteFetchPurgatory (an instance of DelayedOperationPurgatory) to handle timeout, and the request is queued in the remote storage fetcher thread pool task queue. Each thread processes one remote fetch request by…
Acceptance criteria (12)
- A RemoteFetchPurgatory instance is created as a DelayedOperationPurgatory for remote fetch requests
- A remote storage fetcher thread pool is created with size configured by remote.log.reader.threads
- The thread pool task queue has maximum capacity configured by remote.log.reader.max.pending.tasks
- When a fetch request targets an offset in remote storage, it is added to RemoteFetchPurgatory and queued in the thread pool
- Each thread processes one remote fetch request at a time
- The thread finds the RemoteLogSegmentId for the requested offset from RemoteLogMetadataManager
- The thread determines startPosition and endPosition from the offset index
- The thread calls RemoteStorageManager.fetchLogSegmentData(RemoteLogSegmentMetadata, startPosition, endPosition) to retrieve data
- The thread builds a Records instance from the fetched data
- If the remote segment file is deleted, RemoteFetchPurgatory is notified to return an error to the client
- If the remote storage operation fails, the operation is retried with Exponential Back-Off until the original consumer fetch request timeout
- The thread pool enforces the maximum pending tasks limit and rejects or applies backpressure when the queue is full
Clarification questions (2)
- CLARIFICATION NEEDED: Exact Exponential Back-Off parameters (initial delay, multiplier, max delay) not specified
- CLARIFICATION NEEDED: Error response type when remote segment is deleted not specified
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 2. Remote Storage Fetcher Thread Pool", full paragraph starting with "When handling consumer fetch request..."
Why it matches: Remote consumer fetch with purgatory matches the RemoteFetchPurgatory implementation in SR-19. (checked by hand)
2020-05-13 KAFKA-9990 Supporting transactions in tiered storage Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-02-24 KAFKA-12368 Inmemory implementation of RSM and RLMM. Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-03-05 KAFKA-12429 Serdes for all message types in internal topic which is used in default implementation for RLMM. Planned by SprintReady
Sub-task. RLMM default implementation is based on storing all the metadata in an internal topic. We need serdes and format of the message types that need to be stored in the topic. You can see more details in the [KIP|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-MessageFormat]
Define a configuration namespace prefix `remote.log.metadata.*` for RemoteLogMetadataManager implementation-specific configuration properties. These configurations are dependent on the specific metadata storage backend implementation and must be passed through to the configured RemoteLogMetadataManager instance during initialization. The broker must not validate the contents of these properties but must collect and p…
Acceptance criteria (5)
- The broker configuration parser recognizes all properties with prefix `remote.log.metadata.*`
- All properties matching the prefix are collected into a configuration map
- The configuration map is passed to the RemoteLogMetadataManager instance during initialization via its configure() method
- The broker does not attempt to validate the values of these properties
- Properties with this prefix are documented as implementation-dependent
Traced to: KIP-405 Section "Public Interfaces > Configs > RemoteLogMetadataManager", statement "remote.log.metadata.*"
Define the RemoteLogMetadataManager interface as specified in the design document. This interface provides the lifecycle operations for metadata about remote log segments with strongly consistent semantics. The interface must support storing, fetching, and managing remote log segment metadata, and must allow for pluggable custom implementations while providing a default implementation backed by an internal Kafka topi…
Acceptance criteria (8)
- A Java interface named RemoteLogMetadataManager is created with all methods specified in the design document
- The interface includes a putRemoteLogSegmentData method that accepts RemoteLogSegmentId and RemoteLogSegmentMetadata
- The interface includes methods to fetch remote log segment metadata for a given offset
- The interface includes a method to register a topic partition for metadata cache building
- The interface includes an earliestLogOffset method that accepts TopicPartition and returns the earliest offset tracked in remote storage
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > RemoteLogMetadataManager", statement "RemoteLogMetadataManager is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics." and "[code omitted]"
Why it matches: Serdes for metadata topic messages are part of the RemoteLogMetadataManager implementation covered in SR-14. (checked by hand)
2021-03-12 KAFKA-12458 Implementation of Tiered Storage Integration with Azure Storage (ADLS + Blob Storage) Added after Feb 2020
Sub-task. Task to cover integration support for Azure Storage * Azure Blob Storage * Azure Data Lake Store Will split task up later into distinct tracks and components
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-05-17 KAFKA-12802 Add a file based cache for consumed remote log metadata for each partition to avoid consuming again incase of broker restarts. Planned by SprintReady
Sub-task. Add a file based cache for consumed remote log metadata for each partition to avoid consuming again in case of broker restarts.
Implement cache management logic in RemoteLogMetadataManager where leader replicas update the cache by producing metadata to the remote log metadata topic, and follower replicas build and maintain the cache by consuming from the respective remote log metadata topic partitions. The cache must be disk-backed (e.g., RocksDB) to handle potentially large metadata sets without excessive memory usage.
Acceptance criteria (7)
- RemoteLogMetadataManager maintains an in-memory or disk-backed cache of remote log segment metadata
- For leader replicas, after copying a segment to remote storage, RemoteLogManager calls RemoteLogMetadataManager.putRemoteLogSegmentData, which writes the metadata to the remote log metadata topic and updates the local cache
- For follower replicas, RemoteLogMetadataManager subscribes to the relevant remote log metadata topic partitions
- For follower replicas, RemoteLogMetadataManager consumes metadata messages and adds them to the local cache
- When a topic partition is reassigned to a new broker, RemoteLogMetadataManager subscribes to the corresponding remote log metadata topic partition if not already subscribed
- The cache is backed by a persistent store (e.g., RocksDB) to minimize memory footprint
- Committed offsets for consumed metadata topic partitions are stored in a local file to avoid re-reading messages after broker restart
Clarification questions (2)
- CLARIFICATION NEEDED: Exact disk-backed cache implementation (RocksDB confirmed as example, but not mandated)
- CLARIFICATION NEEDED: Local file format and location for storing committed offsets not specified
Traced to: KIP-405 Section "Proposed Changes > RemoteLogMetadataManager implemented with an internal topic", paragraphs starting with "RLMM registers the topic partitions..." and "This requires the cache to be based on disk storage like RocksDB..."
Why it matches: File-based cache for consumed metadata to avoid re-reading after restart is described in SR-31. (checked by hand)
2021-05-20 KAFKA-12816 Add tier storage configs. Planned by SprintReady
Sub-task. Add all the tier storage related configurations including remote log manager, remote storage manager, and remote log metadata manager. These configs are described in the KIP-405 [here|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-Configs.1].
Add a new system-wide boolean configuration property `remote.log.storage.enable` to control whether tiered storage is activated. This property provides backward compatibility for existing Kafka deployments that do not use tiered storage. The default value must be `false` to ensure no impact on clusters not opting into the tiered storage feature. The configuration must be readable at broker startup and govern all remo…
Acceptance criteria (7)
- A new configuration property `remote.log.storage.enable` is added to the broker configuration schema
- Valid values are strictly `true` or `false`
- The default value is `false`
- When set to `false`, no remote log manager components are initialized
- When set to `true`, the broker proceeds to validate and initialize remote storage manager components
- The configuration value is loaded at broker startup and cannot be changed without restart
- The property is documented in the configuration reference with its backward compatibility purpose stated
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.storage.enable - Whether to enable remote log storage or not. Valid values are `true` or `false` and the default value is false. This property gives backward compatibility."
Add a mandatory system-wide configuration property `remote.log.storage.manager.class.name` that specifies the fully qualified class name of the RemoteStorageManager implementation to be used. This configuration is required only when `remote.log.storage.enable` is set to `true`. If `remote.log.storage.enable` is `true` and this property is not configured, the broker must fail startup with a clear error message indicat…
Acceptance criteria (6)
- A new configuration property `remote.log.storage.manager.class.name` is added to the broker configuration schema
- The property accepts a fully qualified Java class name as a string value
- The property is marked as mandatory when `remote.log.storage.enable` is `true`
- If `remote.log.storage.enable` is `true` and `remote.log.storage.manager.class.name` is not set, broker startup fails with error message stating the property is mandatory
- The specified class is validated at startup to ensure it exists on the classpath and implements the RemoteStorageManager interface
- If the class does not exist or does not implement RemoteStorageManager, broker startup fails with a descriptive error
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.storage.manager.class.name - This is mandatory if the remote.log.storage.enable is set as true."
Add an optional system-wide configuration property `remote.log.metadata.manager.class.name` that specifies the fully qualified class name of a custom RemoteLogMetadataManager implementation. If this property is not configured, the broker must use a default inbuilt metadata manager backed by an internal Kafka topic. The configuration must be validated at startup if provided.
Acceptance criteria (7)
- A new configuration property `remote.log.metadata.manager.class.name` is added to the broker configuration schema
- The property is optional
- The property accepts a fully qualified Java class name as a string value
- If the property is not set, the broker initializes the default RemoteLogMetadataManager implementation backed by an internal topic
- If the property is set, the specified class is validated at startup to ensure it exists on the classpath and implements the RemoteLogMetadataManager interface
- If the class does not exist or does not implement RemoteLogMetadataManager, broker startup fails with a descriptive error
- The default internal-topic-based metadata manager is documented as the fallback behavior
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.metadata.manager.class.name(optional) - This is an optional property. If this is not configured, Kafka uses an inbuilt metadata manager backed by an internal topic."
Define a configuration namespace prefix `remote.log.storage.*` for RemoteStorageManager implementation-specific configuration properties. These configurations are dependent on the specific remote storage backend implementation (e.g., HDFS, S3) and must be passed through to the configured RemoteStorageManager instance during initialization. The broker must not validate the contents of these properties but must collect…
Acceptance criteria (5)
- The broker configuration parser recognizes all properties with prefix `remote.log.storage.*`
- All properties matching the prefix are collected into a configuration map
- The configuration map is passed to the RemoteStorageManager instance during initialization via its configure() method
- The broker does not attempt to validate the values of these properties
- Properties with this prefix are documented as implementation-dependent
Traced to: KIP-405 Section "Public Interfaces > Configs > RemoteStorageManager", statement "remote.log.storage.*"
Define a configuration namespace prefix `remote.log.metadata.*` for RemoteLogMetadataManager implementation-specific configuration properties. These configurations are dependent on the specific metadata storage backend implementation and must be passed through to the configured RemoteLogMetadataManager instance during initialization. The broker must not validate the contents of these properties but must collect and p…
Acceptance criteria (5)
- The broker configuration parser recognizes all properties with prefix `remote.log.metadata.*`
- All properties matching the prefix are collected into a configuration map
- The configuration map is passed to the RemoteLogMetadataManager instance during initialization via its configure() method
- The broker does not attempt to validate the values of these properties
- Properties with this prefix are documented as implementation-dependent
Traced to: KIP-405 Section "Public Interfaces > Configs > RemoteLogMetadataManager", statement "remote.log.metadata.*"
Add a system-wide configuration property `remote.log.manager.thread.pool.size` that specifies the number of threads in the remote log manager thread pool. This thread pool is used for scheduling tasks to copy log segments to remote storage, fetch remote log indexes, and clean up remote log segments. The configuration must accept a positive integer value.
Acceptance criteria (5)
- A new configuration property `remote.log.manager.thread.pool.size` is added to the broker configuration schema
- The property accepts a positive integer value
- The property is validated at startup to reject zero, negative, or non-integer values
- The thread pool is initialized with the configured number of threads
- Each thread is capable of processing topic-partition tasks for copy, fetch, and cleanup operations
Clarification questions (2)
- CLARIFICATION NEEDED: Default value for remote.log.manager.thread.pool.size not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.manager.thread.pool.size Remote log thread pool size, which is used in scheduling tasks to copy segments, fetch remote log indexes and clean up remote log segments."
Add a system-wide configuration property `remote.log.manager.task.interval.ms` that specifies the interval in milliseconds at which the remote log manager runs scheduled tasks such as copying segments to remote storage, fetching remote log indexes, and cleaning up remote log segments. The configuration must accept a positive integer value representing milliseconds.
Acceptance criteria (5)
- A new configuration property `remote.log.manager.task.interval.ms` is added to the broker configuration schema
- The property accepts a positive integer value in milliseconds
- The property is validated at startup to reject zero, negative, or non-integer values
- The remote log manager scheduler uses this interval to schedule topic-partition processing tasks
- After a topic-partition is successfully processed, its next scheduled processing time is set to (current_time + remote.log.manager.task.interval.ms)
Clarification questions (2)
- CLARIFICATION NEEDED: Default value for remote.log.manager.task.interval.ms not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.manager.task.interval.ms The interval at which remote log manager runs the scheduled tasks like copy segments, fetch remote log indexes and clean up remote log segments."
Add a system-wide configuration property `remote.log.reader.threads` that specifies the number of threads in the remote log reader thread pool. This thread pool handles consumer fetch requests that require reading data from remote storage. The configuration must accept a positive integer value.
Acceptance criteria (5)
- A new configuration property `remote.log.reader.threads` is added to the broker configuration schema
- The property accepts a positive integer value
- The property is validated at startup to reject zero, negative, or non-integer values
- The remote storage fetcher thread pool is initialized with the configured number of threads
- Each thread processes one remote fetch request at a time
Clarification questions (2)
- CLARIFICATION NEEDED: Default value for remote.log.reader.threads not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.reader.threads Remote log reader thread pool size"
Add a system-wide configuration property `remote.log.reader.max.pending.tasks` that specifies the maximum size of the remote log reader thread pool task queue. If the task queue is full, the broker must stop accepting new remote log read requests. The configuration must accept a positive integer value.
Acceptance criteria (6)
- A new configuration property `remote.log.reader.max.pending.tasks` is added to the broker configuration schema
- The property accepts a positive integer value
- The property is validated at startup to reject zero, negative, or non-integer values
- The remote storage fetcher thread pool task queue is initialized with the configured maximum capacity
- When the task queue is full, the broker rejects new remote fetch requests or applies appropriate backpressure
- Rejected requests are handled with a defined error response to the client
Clarification questions (3)
- CLARIFICATION NEEDED: Default value for remote.log.reader.max.pending.tasks not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
- CLARIFICATION NEEDED: Exact error response or backpressure mechanism when queue is full not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.reader.max.pending.tasks Maximum remote log reader thread pool task queue size. If the task queue is full, broker will stop reading remote log segments."
Add a per-topic configuration property `remote.log.retention.minutes` that specifies the retention period in minutes for remote log segments. This value defines how long log segments are retained in remote storage before being eligible for deletion. The configuration must accept a positive integer value representing minutes.
Acceptance criteria (5)
- A new topic-level configuration property `remote.log.retention.minutes` is added to the topic configuration schema
- The property accepts a positive integer value in minutes
- The property can be set at topic creation or via alter-config operations
- The retention calculation for remote log segments uses this value to determine eligibility for deletion
- If not set, the broker must have a defined default behavior or require the value to be set explicitly when tiered storage is enabled
Clarification questions (3)
- CLARIFICATION NEEDED: Default value for remote.log.retention.minutes not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
- CLARIFICATION NEEDED: Interaction or relationship with local retention (log.retention) not fully specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Per Topic Configuration", statement "remote.log.retention.minutes"
Add a per-topic configuration property `remote.log.retention.bytes` that specifies the retention limit in bytes for remote log segments. This value defines the maximum total size of log segments retained in remote storage for a topic-partition before older segments are eligible for deletion. The configuration must accept a positive integer value representing bytes.
Acceptance criteria (5)
- A new topic-level configuration property `remote.log.retention.bytes` is added to the topic configuration schema
- The property accepts a positive integer value in bytes
- The property can be set at topic creation or via alter-config operations
- The retention calculation for remote log segments uses this value to determine eligibility for deletion based on cumulative size
- If not set, the broker must have a defined default behavior or require the value to be set explicitly when tiered storage is enabled
Clarification questions (3)
- CLARIFICATION NEEDED: Default value for remote.log.retention.bytes not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
- CLARIFICATION NEEDED: Interaction or relationship with remote.log.retention.minutes not specified (e.g., OR vs AND logic)
Traced to: KIP-405 Section "Public Interfaces > Configs > Per Topic Configuration", statement "remote.log.retention.bytes"
Why it matches: Adding all tiered storage configurations matches the comprehensive config tickets SR-1 through SR-11. (checked by hand)
2021-06-18 KAFKA-12969 Add cluster or broker level config for topic level tiered storage confgs. Added after Feb 2020
Sub-task. We should have cluster level settings for remote storage configs such as retention settings. This is essential for customers who have hundreds of topic in a cluster don't want to set these TS related config individualy for each topic.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-06-18 KAFKA-12970 Make tiered storage related schemas adopt flexible versions feature. Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-06-24 KAFKA-12988 Change RLMM add/updateRemoteLogSegmentMetadata and putRemotePartitionDeleteMetadata APIS asynchronous. Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-07-16 KAFKA-13097 Handle the requests gracefully to publish the events in TopicBasedRemoteLogMetadataManager when it is not yet initialized. Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2021-10-12 KAFKA-13369 Follower fetch protocol enhancements for tiered storage. Planned by SprintReady
Sub-task.
Modify follower fetch request handling so that leaders only return data that is still in the leader's local storage. Log segments that exist only in remote storage are not replicated to followers, as they are already present in remote storage. Followers retrieve information about remote segments from RemoteLogMetadataManager instead of replicating the data.
Acceptance criteria (5)
- For follower fetch requests, the leader identifies the requested offset range
- The leader serves data only from local log segments
- If the requested offset range includes offsets that exist only in remote storage, those offsets are not included in the fetch response
- Followers are expected to retrieve remote segment metadata from RemoteLogMetadataManager to learn about remote segments
- If a follower becomes a leader, it can locate and serve data from remote storage using the metadata it has retrieved
Clarification questions (1)
- CLARIFICATION NEEDED: Exact mechanism for followers to know they should not request remote-only offsets not fully specified (is it implicit in the offset range or explicit signaling?)
Traced to: KIP-405 Section "Proposed Changes > Follower Requests/Replication", paragraph starting with "For follower fetch, the leader only returns the data..."
Add a new target timestamp value NEXT_LOCAL_TIMESTAMP (value -3) to the ListOffsets API. This value is used by follower replicas to request the offset that they should start fetching to replicate local logs, where all records earlier than this offset can be considered as copied to remote storage. The request and response schemas do not add new fields, but the version must be bumped to indicate support for the new tim…
Acceptance criteria (6)
- ListOffsets API supports a new target timestamp value: NEXT_LOCAL_TIMESTAMP with value -3
- The ListOffsets request and response schemas do not add new fields
- The ListOffsets request and response version is incremented to indicate support for NEXT_LOCAL_TIMESTAMP
- When a follower sends a request with target timestamp = -3 (NEXT_LOCAL_TIMESTAMP), the leader returns the first offset in the local log (i.e., the earliest offset not yet copied to remote storage or the first offset that followers must replicate)
- The returned offset allows followers to avoid fetching records that are already in remote storage
- The implementation documents the use case: followers use this to determine the starting offset for replication
Clarification questions (1)
- CLARIFICATION NEEDED: Exact offset calculation for NEXT_LOCAL_TIMESTAMP not fully specified (is it remote_log_end_offset + 1, or local_log_start_offset, or something else?)
Traced to: KIP-405 Section "Proposed Changes > Other APIs > ListOffsets", paragraph starting with "This API is enhanced with supporting new target timestamp value as -3..."
Why it matches: Follower fetch protocol enhancements match SR-24 and SR-27 for handling remote-only segments and NEXT_LOCAL_TIMESTAMP. (checked by hand)
2022-07-01 KAFKA-14038 Optimize calculation of size for log in remote tier Added after Feb 2020
Improvement. {color:#24292f}As per the Tiered Storage feature introduced in [KIP-405|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage], users can configure the retention of remote tier based on time, by size, or both. The work of computing the log segments to be deleted based on the retention config is [owned by RemoteLogManager|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Ti…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2022-12-13 KAFKA-14466 Refactor ClassLoaderAwareRemoteStorageManager.scala to ClassLoaderAwareRemoteStorageManager.java and move it to storage module. Added after Feb 2020
Sub-task. https://github.com/apache/kafka/pull/11390#discussion_r1043982906
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2022-12-14 KAFKA-14469 Fix inconsistency in config name between code and KIP-405 Added after Feb 2020
Sub-task. We have 3 tiered storage configuration keys that are not adhering to KIP-405 specification. KIP-405 has ".log" in the keys. public static final String REMOTE_LOG_STORAGE_ENABLE_CONFIG = "remote.storage.enable"; public static final String LOCAL_LOG_RETENTION_MS_CONFIG = "local.retention.ms"; public static final String LOCAL_LOG_RETENTION_BYTES_CONFIG = "local.retention.bytes"; Reference: KIP-405: [https://cwi…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2022-12-15 KAFKA-14495 Improve the RemoteIndexCacheTest Added after Feb 2020
Sub-task. https://github.com/apache/kafka/pull/11390/files#r1049392445
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-02-06 KAFKA-14685 TierStateMachine interface for building remote aux log Added after Feb 2020
Sub-task. To help with https://issues.apache.org/jira/browse/KAFKA-13560 , we can introduce an interface to manage state transitions of building the remote aux log asynchronously
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-03-27 KAFKA-14850 Introduce InMemoryLeaderEpochCheckpoint to allow quick write/read Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-04-11 KAFKA-14888 RemoteLogManager - deleting expired/size breached log segments to remote storage implementation Planned by SprintReady
Sub-task. Implementation of RLM as mentioned in the HLD section of KIP-405, this JIRA covers deleting time/size breached log segments in remote storage. [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP-405:KafkaTieredStorage-High-leveldesign]
Implement the RLM task that handles expiration and deletion of remote log segments based on remote retention configuration. The leader computes segments to be deleted, updates the earliest offset in RemoteLogMetadataManager, removes segment metadata and data from remote storage, and handles failures by logging them to a dedicated topic. Followers fetch the earliest offset and update their local indexes and start offs…
Acceptance criteria (9)
- For leader replicas, the task computes expired remote log segments based on remote.log.retention.minutes and remote.log.retention.bytes
- The task updates the earliest offset for the topic partition in RemoteLogMetadataManager
- The task retrieves all remote log segment IDs eligible for deletion
- The task calls RemoteStorageManager.deleteLogSegment for each segment ID
- The task removes metadata entries via RemoteLogMetadataManager
- If deletion from remote storage fails, the failure is logged with the RemoteLogSegmentId to a dedicated topic (default: __remote_segments_to_be_deleted)
- For follower replicas, the task calls RemoteLogMetadataManager.earliestLogOffset(TopicPartition) to fetch the earliest offset
- Followers clean up local indexes up to the received earliest offset
- Followers update their local log start offset with the received value
Clarification questions (2)
- CLARIFICATION NEEDED: Exact name and schema of the __remote_segments_to_be_deleted topic not fully specified
- CLARIFICATION NEEDED: Cleanup mechanism for entries in __remote_segments_to_be_deleted topic not specified
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 1. Remote Log Manager (RLM) Thread Pool > Handle expired remote segments (leader and follower)", full paragraph starting with "RLM leader computes the log segments to be deleted..."
Why it matches: Deleting expired/size-breached remote log segments matches the RLM task in SR-17. (checked by hand)
2023-04-11 KAFKA-14889 RemoteLogManager - allow consumer fetch records from remote storage implementation Planned by SprintReady
Sub-task. Implementation of RLM as mentioned in the HLD section of KIP-405, this JIRA covers enabling consumers fetch records from remote storage [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP-405:KafkaTieredStorage-High-leveldesign] h4.
Implement the remote storage fetcher thread pool that processes consumer fetch requests requiring data from remote storage. When a fetch request targets an offset in remote storage, the request is added to a RemoteFetchPurgatory (an instance of DelayedOperationPurgatory) to handle timeout, and the request is queued in the remote storage fetcher thread pool task queue. Each thread processes one remote fetch request by…
Acceptance criteria (12)
- A RemoteFetchPurgatory instance is created as a DelayedOperationPurgatory for remote fetch requests
- A remote storage fetcher thread pool is created with size configured by remote.log.reader.threads
- The thread pool task queue has maximum capacity configured by remote.log.reader.max.pending.tasks
- When a fetch request targets an offset in remote storage, it is added to RemoteFetchPurgatory and queued in the thread pool
- Each thread processes one remote fetch request at a time
- The thread finds the RemoteLogSegmentId for the requested offset from RemoteLogMetadataManager
- The thread determines startPosition and endPosition from the offset index
- The thread calls RemoteStorageManager.fetchLogSegmentData(RemoteLogSegmentMetadata, startPosition, endPosition) to retrieve data
- The thread builds a Records instance from the fetched data
- If the remote segment file is deleted, RemoteFetchPurgatory is notified to return an error to the client
- If the remote storage operation fails, the operation is retried with Exponential Back-Off until the original consumer fetch request timeout
- The thread pool enforces the maximum pending tasks limit and rejects or applies backpressure when the queue is full
Clarification questions (2)
- CLARIFICATION NEEDED: Exact Exponential Back-Off parameters (initial delay, multiplier, max delay) not specified
- CLARIFICATION NEEDED: Error response type when remote segment is deleted not specified
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 2. Remote Storage Fetcher Thread Pool", full paragraph starting with "When handling consumer fetch request..."
Modify consumer fetch request handling in ReplicaManager to delegate reads to RemoteLogManager when the requested offset is not in local storage. For any fetch request, ReplicaManager calls readFromLocalLog; if this method returns an OffsetOutOfRangeException, ReplicaManager delegates the read call to RemoteLogManager to serve the data from remote storage.
Acceptance criteria (6)
- ReplicaManager fetch request handling first attempts to read from local log via readFromLocalLog
- If readFromLocalLog succeeds, the data is returned to the client as normal
- If readFromLocalLog throws OffsetOutOfRangeException, ReplicaManager checks if the offset is within the remote log offset range
- If the offset is in the remote log range, ReplicaManager delegates the read to RemoteLogManager
- RemoteLogManager handles the remote fetch as per the remote storage fetcher thread pool logic
- If the offset is not in the remote log range (i.e., before remote log start offset or after log end offset), an appropriate error is returned to the client
Traced to: KIP-405 Section "Proposed Changes > Consumer Fetch Requests", paragraph starting with "For any fetch requests, ReplicaManager will proceed..."
Why it matches: Allowing consumer fetch from remote storage matches SR-19 and SR-23 for remote fetch handling. (checked by hand)
2023-05-02 KAFKA-14953 Add metrics for tiered storage Added after Feb 2020
Sub-task. Not just for expired fetch. We also need to add all the metrics described in KIP-405 https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-NewMetrics ref: [https://github.com/apache/kafka/pull/13535#discussion_r1180286031]
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-05-12 KAFKA-14993 Improve TransactionIndex instance handling while copying to and fetching from RSM. Added after Feb 2020
Improvement. RSM should throw a ResourceNotFoundException if it does not have TransactionIndex. Currently, it expects an empty InputStream and creates an unnecessary file in the cache. This can be avoided by catching ResourceNotFoundException and not creating an instance. There are minor cleanups needed in RemoteIndexCache and other TransactionIndex usages. Also, update the LocalTieredStorage, see [this|https://github.com/apac…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-05-30 KAFKA-15037 initialize unifiedLog with remoteStorageSystemEnable correctly Added after Feb 2020
Sub-task. UnifiedLog relied on the `remoteStorageSystemEnable` to identify if the broker is enabling remote storage, but we never pass this value from the config into UnifiedLog. So it'll always be false.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-05-31 KAFKA-15040 segment copy to remote storage won't work in KRaft mode Added after Feb 2020
Sub-task. Currently, when received LeaderAndIsr request, we'll notify remoteLogManager about this leadership changed to trigger the following workflow. But LeaderAndIsr won't be sent in KRaft mode, instead, the topicDelta will be received.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-06-07 KAFKA-15066 passing listener name config into TopicBasedRemoteLogMetadataManagerConfig Added after Feb 2020
Sub-task. The `remote.log.metadata.manager.listener.name` config doesn't pass to TopicBasedRemoteLogMetadataManagerConfig correctly.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-06-13 KAFKA-15083 Passing "remote.log.metadata.*" configs into RLMM Planned by SprintReady
Sub-task. Based on the [KIP-405|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-Configs]: |_{color:#000000}remote.log.metadata.*{color}_|{color:#000000}Default RLMM implementation creates producer and consumer instances. Common client propoerties can be configured with `remote.log.metadata.common.client.` prefix. User can also pass properties specific to {color}{col…
Define a configuration namespace prefix `remote.log.metadata.*` for RemoteLogMetadataManager implementation-specific configuration properties. These configurations are dependent on the specific metadata storage backend implementation and must be passed through to the configured RemoteLogMetadataManager instance during initialization. The broker must not validate the contents of these properties but must collect and p…
Acceptance criteria (5)
- The broker configuration parser recognizes all properties with prefix `remote.log.metadata.*`
- All properties matching the prefix are collected into a configuration map
- The configuration map is passed to the RemoteLogMetadataManager instance during initialization via its configure() method
- The broker does not attempt to validate the values of these properties
- Properties with this prefix are documented as implementation-dependent
Traced to: KIP-405 Section "Public Interfaces > Configs > RemoteLogMetadataManager", statement "remote.log.metadata.*"
Why it matches: Passing remote.log.metadata.* configs to RLMM matches the configuration namespace in SR-5. (checked by hand)
2023-06-13 KAFKA-15084 Remove lock contention in RemoteIndexCache Added after Feb 2020
Sub-task. RemoteIndexCache cache is accessed from multiple threads concurrently in the fetch from consumer code path [1]. Currently, the RemoteIndexCache uses LinkedHashMap as the cache implementation internally. Since LinkedHashMap is not a thread safe data structure, we use coarse grained lock on the entire map/cache when writing to the cache. This means that if a thread if fetching information from a particular segme…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-06-28 KAFKA-15130 Delete remote segments when delete a topic Added after Feb 2020
Improvement. When tiered storage is enabled and {{delete.topic.enable=true}} , deleting a topic should also delete the corresponding segments of that topic on the remote system, and cancel the RLMTask for that topic.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-06-28 KAFKA-15131 Improve RemoteStorageManager exception handling documentation Added after Feb 2020
Improvement. As discussed here[1], RemoteStorageManager javadocs requires clarification regarding error handling: * Remove ambiguity on `RemoteResourceNotFoundException` description * Describe when `RemoteResourceNotFoundException` can/should be thrown * Describe expectations for idempotent operations when copying/deleting [1] https://issues.apache.org/jira/browse/KAFKA-7739?focusedCommentId=17720936&page=com.atlass…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-06-28 KAFKA-15132 Implement disable & re-enablement for Tiered Storage Added after Feb 2020
New Feature. KIP-405 [1] introduces the Tiered Storage feature in Apache Kafka. One of the limitations mentioned in the KIP is inability to re-enable TS on a topic after it has been disabled. {quote}Once tier storage is enabled for a topic, it can not be disabled. We will add this feature in future versions. One possible workaround is to create a new topic and copy the data from the desired offset and delete the old topic. …
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-06-30 KAFKA-15135 RLM listener configurations passed but ignored by RLMM Added after Feb 2020
Bug. As describe here [1] properties captured from listener are passed but ignored by TopicBasedRLMM. [1] https://github.com/apache/kafka/pull/13828#issuecomment-1611155345
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-06 KAFKA-15157 Print startup time for RemoteIndexCache Added after Feb 2020
Sub-task. When RemoteIndexCache starts up, it will try to re-build the in-memory cache using the files already present on the disk in the remote-index-cache directory. The process involves: 1. deleting existing files which are pending delete i.e. have a .delete suffix 2. read the cached index files, if present. 3. creating the indexes (this step will create a MMapped'buffer) 4. perform sanity check on the indexes 5. add t…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-08 KAFKA-15167 Tiered Storage Test Harness Framework Added after Feb 2020
Sub-task. Base class for integration tests exercising the tiered storage functionality in Kafka.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-08 KAFKA-15168 Handle overlapping remote log segments in RemoteLogMetadata cache Added after Feb 2020
Sub-task. For a partition p0 and a given leader epoch, the remote log manager can upload duplicate segments due to leader change. RemoteLogMetadata cache should handle the duplicate segments which may affect the follower and consumer. (eg) L0 uploaded the segment PZyYVdsJQWeBAdBDPqkcVA at t0 for offset range 10 - 90 L1 uploads the segment L5Ufv71IToiZYKgsluzcyA at t1 for offset range 5 - 100 In the [RemoteLogLeaderEp…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-11 KAFKA-15176 Add missing tests for remote storage metrics Added after Feb 2020
Sub-task. {{RemoteLogReaderTaskQueueSize}} {{RemoteLogReaderAvgIdlePercent}} {{RemoteLogManagerTasksAvgIdlePercent}} {{}} https://github.com/apache/kafka/pull/13944#pullrequestreview-1513943273{{{}{}}}
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-12 KAFKA-15181 Race condition on partition assigned to TopicBasedRemoteLogMetadataManager Added after Feb 2020
Sub-task. TopicBasedRemoteLogMetadataManager (TBRLMM) uses a cache to be prepared whever partitions are assigned. When partitions are assigned to the TBRLMM instance, a consumer is started to keep the cache up to date. If the cache hasn't finalized to build, TBRLMM fails to return remote metadata about partitions that are store on the backing topic. TBRLMM may not recover from this failing state. A proposal to fix thi…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-14 KAFKA-15189 Do not initialize RemoteStorage related metrics when disabled at cluster Added after Feb 2020
Bug. context: https://github.com/apache/kafka/pull/13944#discussion_r1251099820
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-17 KAFKA-15194 Rename local tiered storage segment with offset as prefix for easy navigation Added after Feb 2020
Task. In LocalTieredStorage which is an implementation of RemoteStorageManager, segments are saved with random UUID. This makes navigating to a particular segment harder. To navigate a given segment by offset, prepend the offset information to the segment filename. https://github.com/apache/kafka/pull/13837#discussion_r1258896009
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-22 KAFKA-15236 Rename Remote Storage metrics to remove ambiguity Added after Feb 2020
Sub-task. As per the Tiered Storage feature introduced in [KIP-405|https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage], we added several metrics related to reads(from) and writes(to) for remote storage. The naming convention that was followed is confusing to the users. For eg. in regular Kafka, BytesIn means bytes *_written_* to the log, and BytesOut means bytes *_read_* from the log. But with…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-27 KAFKA-15260 RLM Task should wait until RLMM is initialized before copying segments to remote Added after Feb 2020
Sub-task. The RLM Task uploads segment to the remote storage for its leader partitions and after each upload it sends a message 'COPY_SEGMENT_STARTED' to the Topic based RLMM topic and then waits for the TBRLMM to consume the message before continuing. If the RLMM is not initialized, TBRLMM may not be able to consume the message within the stipulated time and timeout and RLMM will repeat later. It make take a few mins for t…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-27 KAFKA-15261 ReplicaFetcher thread should not block if RLMM is not initialized Added after Feb 2020
Sub-task. While building remote log aux state, the replica fetcher fetches the remote log segment metadata. If the TBRLMM is not initialized yet, the call blocks. Since replica fetchers share a common lock, it prevents other replica fetchers from running as well. Also the same lock is shared in the handle LeaderAndISR request path, hence those calls get blocked as well. Instead, replica fetcher should check if RLMM is initi…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-28 KAFKA-15267 Cluster-wide disablement of Tiered Storage Added after Feb 2020
Sub-task. h2. Summary KIP-405 defines the configuration {{remote.log.storage.system.enable}} which controls whether all resources needed for Tiered Storage to function are instantiated properly in Kafka. However, the interaction between remote data and Kafka if that configuration is set to false while there are still topics with {{{}remote.storage.enable is undefined{}}}. {color:#ff8b00}*We would like to give customers the …
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-07-31 KAFKA-15274 support moving files to be deleted to other directories Added after Feb 2020
Task. Hello everyone, I am a Kafka user from China. Our company operates in public clouds overseas, such as AWS, Ali Cloud, and Huawei Cloud. We face a large amount of data exchange and business message delivery every day. Daily messages consume a significant amount of disk space. Purchasing the corresponding storage capacity on these cloud providers incurs substantial costs, especially for SSDs with ultra-high IOPS. High …
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-01 KAFKA-15290 Add support to onboard existing topics to tiered storage Added after Feb 2020
Sub-task. This task is about adding support to enable tiered storage for existing topics in the cluster.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-02 KAFKA-15293 Update metrics doc to add tiered storage metrics Added after Feb 2020
Sub-task.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-02 KAFKA-15294 Make remote storage related configs as public (i.e. non-internal) Added after Feb 2020
Sub-task. We should publish all the remote storage related configs in v3.6.0. It can be verified by: [code omitted] {{}}
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-02 KAFKA-15295 Add config validation when remote storage is enabled on a topic Added after Feb 2020
Sub-task. If system level remote storage is not enabled, then enabling remote storage on a topic should throw exception while validating the configs. See https://github.com/apache/kafka/pull/14114#discussion_r1280372441 for more details
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-02 KAFKA-15298 Disable DeleteRecords on Tiered Storage topics Closed as Won't Fix
Sub-task. Currently the DeleteRecords API does not work with Tiered Storage. We should ensure that this is reflected in the responses that clients get when trying to use the API with tiered topics.
Closed as Won't Fix: DeleteRecords already worked (SprintReady had specified it in SR-25).
2023-08-10 KAFKA-15329 Make default `remote.log.metadata.manager.class.name` as topic based RLMM Planned by SprintReady
Sub-task. As described in [https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-Configs.1] , we should set default "remote.log.metadata.manager.class.name" as topic based RLMM.
Add an optional system-wide configuration property `remote.log.metadata.manager.class.name` that specifies the fully qualified class name of a custom RemoteLogMetadataManager implementation. If this property is not configured, the broker must use a default inbuilt metadata manager backed by an internal Kafka topic. The configuration must be validated at startup if provided.
Acceptance criteria (7)
- A new configuration property `remote.log.metadata.manager.class.name` is added to the broker configuration schema
- The property is optional
- The property accepts a fully qualified Java class name as a string value
- If the property is not set, the broker initializes the default RemoteLogMetadataManager implementation backed by an internal topic
- If the property is set, the specified class is validated at startup to ensure it exists on the classpath and implements the RemoteLogMetadataManager interface
- If the class does not exist or does not implement RemoteLogMetadataManager, broker startup fails with a descriptive error
- The default internal-topic-based metadata manager is documented as the fallback behavior
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.metadata.manager.class.name(optional) - This is an optional property. If this is not configured, Kafka uses an inbuilt metadata manager backed by an internal topic."
Why it matches: Setting default remote.log.metadata.manager.class.name to topic-based RLMM matches SR-3. (checked by hand)
2023-08-16 KAFKA-15351 Update log-start-offset after leader election for topics enabled with remote storage Added after Feb 2020
Sub-task. Case-1: In the FETCH response, the leader-log-start-offset will be piggy-backed. But, there can be a scenario: # Leader deleted the remote log segment and updates it's log-start-offset # Before the replica-2 update it's log-start-offset via FETCH-request, the leadership changed to replica-2. # There are no more eligible segments to delete from remote. # The log-start-offset will be stale (referring to old …
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-16 KAFKA-15352 Ensure consistency while deleting the remote log segments Added after Feb 2020
Sub-task. In Kafka-14888, the remote log segments are deleted which breaches the retention time/size before updating the log-start-offset. In middle of deletion, if the consumer starts to read from the beginning of the topic, then it will fail to read the messages and UNKNOWN_SERVER_ERROR will be thrown back to the consumer. To ensure consistency, similar to local log segments where the actual segments are deleted after {{{…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-21 KAFKA-15386 Update log-start-offset in RLM task for a leader partition for the earliest leader epoch of the current leader's leader epoch lineage available in the remote storage. Added after Feb 2020
Task. Update log-start-offset in RLM task for a leader partition for the earliest leader epoch of the current leader's leader epoch lineage available in the remote storage.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-24 KAFKA-15399 Enable OffloadAndConsumeFromLeader test Added after Feb 2020
Sub-task. Build / JDK 17 and Scala 2.13 / initializationError – org.apache.kafka.tiered.storage.integration.OffloadAndConsumeFromLeaderTest
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-28 KAFKA-15410 Add functional integration tests with tiered storage Added after Feb 2020
Sub-task. Add the functional integration tests with tiered storage: # PartitionsExpandTest # DeleteTopicTest # DeleteSegmentsByRetentionSizeTest # DeleteSegmentsByRetentionTimeTest # DeleteSegmentsDueToLogStartOffsetBreachTest # EnableRemoteLogOnTopicTest # ReassignReplicaExpandTest # ReassignReplicaMoveTest and # ReassignReplicaShrinkTest
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-08-29 KAFKA-15414 remote logs get deleted after partition reassignment Added after Feb 2020
Bug. it seems I'm reaching that codepath when running reassignments on my cluster and segment are deleted from remote store despite a huge retention (topic created a few hours ago with 1000h retention). It seems to happen consistently on some partitions when reassigning but not all partitions. My test: I have a test topic with 30 partition configured with 1000h global retention and 2 minutes local retention I have…
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-09-01 KAFKA-15427 Integration tests in TS test harness detect resource leaks Added after Feb 2020
Sub-task. The pull request ([https://github.com/apache/kafka/pull/14116]) for adding the Tiered Storage test harness uncovered resource leaks as part of the build ([https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14116/28/testReport/junit/org.apache.kafka.tiered.storage.integration/OffloadAndConsumeFromLeaderTest/Build___JDK_20_and_Scala_2_13___initializationError/)] This can be reproduced locally by running the …
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
2023-09-07 KAFKA-15442 add document to introduce tiered storage feature and the usage Added after Feb 2020
Sub-task. Add a section in the document to introduce the tiered storage feature and how to enable it, use it.
Work the February 2020 document didn't describe: added later (new scope, tests, metrics, KRaft support, or a fix found in 2023).
Detailed view: all 38 tickets SprintReady wrote
Each one in full: description, acceptance criteria, clarification questions and the section it's traced to.
SR-1 Implement System-Wide Configuration for Remote Log Storage Enable FlagMatches KAFKA-12816
Add a new system-wide boolean configuration property `remote.log.storage.enable` to control whether tiered storage is activated. This property provides backward compatibility for existing Kafka deployments that do not use tiered storage. The default value must be `false` to ensure no impact on clusters not opting into the tiered storage feature. The configuration must be readable at broker startup and govern all remo…
Acceptance criteria (7)
- A new configuration property `remote.log.storage.enable` is added to the broker configuration schema
- Valid values are strictly `true` or `false`
- The default value is `false`
- When set to `false`, no remote log manager components are initialized
- When set to `true`, the broker proceeds to validate and initialize remote storage manager components
- The configuration value is loaded at broker startup and cannot be changed without restart
- The property is documented in the configuration reference with its backward compatibility purpose stated
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.storage.enable - Whether to enable remote log storage or not. Valid values are `true` or `false` and the default value is false. This property gives backward compatibility."
SR-2 Implement Mandatory Configuration for Remote Storage Manager Class NameMatches KAFKA-12816
Add a mandatory system-wide configuration property `remote.log.storage.manager.class.name` that specifies the fully qualified class name of the RemoteStorageManager implementation to be used. This configuration is required only when `remote.log.storage.enable` is set to `true`. If `remote.log.storage.enable` is `true` and this property is not configured, the broker must fail startup with a clear error message indicat…
Acceptance criteria (6)
- A new configuration property `remote.log.storage.manager.class.name` is added to the broker configuration schema
- The property accepts a fully qualified Java class name as a string value
- The property is marked as mandatory when `remote.log.storage.enable` is `true`
- If `remote.log.storage.enable` is `true` and `remote.log.storage.manager.class.name` is not set, broker startup fails with error message stating the property is mandatory
- The specified class is validated at startup to ensure it exists on the classpath and implements the RemoteStorageManager interface
- If the class does not exist or does not implement RemoteStorageManager, broker startup fails with a descriptive error
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.storage.manager.class.name - This is mandatory if the remote.log.storage.enable is set as true."
SR-3 Implement Optional Configuration for Remote Log Metadata Manager Class NameMatches KAFKA-9555 KAFKA-12816 KAFKA-15329
Add an optional system-wide configuration property `remote.log.metadata.manager.class.name` that specifies the fully qualified class name of a custom RemoteLogMetadataManager implementation. If this property is not configured, the broker must use a default inbuilt metadata manager backed by an internal Kafka topic. The configuration must be validated at startup if provided.
Acceptance criteria (7)
- A new configuration property `remote.log.metadata.manager.class.name` is added to the broker configuration schema
- The property is optional
- The property accepts a fully qualified Java class name as a string value
- If the property is not set, the broker initializes the default RemoteLogMetadataManager implementation backed by an internal topic
- If the property is set, the specified class is validated at startup to ensure it exists on the classpath and implements the RemoteLogMetadataManager interface
- If the class does not exist or does not implement RemoteLogMetadataManager, broker startup fails with a descriptive error
- The default internal-topic-based metadata manager is documented as the fallback behavior
Traced to: KIP-405 Section "Public Interfaces > Configs > System-Wide", statement "remote.log.metadata.manager.class.name(optional) - This is an optional property. If this is not configured, Kafka uses an inbuilt metadata manager backed by an internal topic."
SR-4 Implement Remote Storage Manager-Specific Configuration NamespaceMatches KAFKA-12816 KAFKA-9565 KAFKA-9569 KAFKA-12458
Define a configuration namespace prefix `remote.log.storage.*` for RemoteStorageManager implementation-specific configuration properties. These configurations are dependent on the specific remote storage backend implementation (e.g., HDFS, S3) and must be passed through to the configured RemoteStorageManager instance during initialization. The broker must not validate the contents of these properties but must collect…
Acceptance criteria (5)
- The broker configuration parser recognizes all properties with prefix `remote.log.storage.*`
- All properties matching the prefix are collected into a configuration map
- The configuration map is passed to the RemoteStorageManager instance during initialization via its configure() method
- The broker does not attempt to validate the values of these properties
- Properties with this prefix are documented as implementation-dependent
Traced to: KIP-405 Section "Public Interfaces > Configs > RemoteStorageManager", statement "remote.log.storage.*"
SR-5 Implement Remote Log Metadata Manager-Specific Configuration NamespaceMatches KAFKA-12429 KAFKA-12816 KAFKA-15083
Define a configuration namespace prefix `remote.log.metadata.*` for RemoteLogMetadataManager implementation-specific configuration properties. These configurations are dependent on the specific metadata storage backend implementation and must be passed through to the configured RemoteLogMetadataManager instance during initialization. The broker must not validate the contents of these properties but must collect and p…
Acceptance criteria (5)
- The broker configuration parser recognizes all properties with prefix `remote.log.metadata.*`
- All properties matching the prefix are collected into a configuration map
- The configuration map is passed to the RemoteLogMetadataManager instance during initialization via its configure() method
- The broker does not attempt to validate the values of these properties
- Properties with this prefix are documented as implementation-dependent
Traced to: KIP-405 Section "Public Interfaces > Configs > RemoteLogMetadataManager", statement "remote.log.metadata.*"
SR-6 Implement Remote Log Manager Thread Pool Size ConfigurationMatches KAFKA-12816
Add a system-wide configuration property `remote.log.manager.thread.pool.size` that specifies the number of threads in the remote log manager thread pool. This thread pool is used for scheduling tasks to copy log segments to remote storage, fetch remote log indexes, and clean up remote log segments. The configuration must accept a positive integer value.
Acceptance criteria (5)
- A new configuration property `remote.log.manager.thread.pool.size` is added to the broker configuration schema
- The property accepts a positive integer value
- The property is validated at startup to reject zero, negative, or non-integer values
- The thread pool is initialized with the configured number of threads
- Each thread is capable of processing topic-partition tasks for copy, fetch, and cleanup operations
Clarification questions (2)
- CLARIFICATION NEEDED: Default value for remote.log.manager.thread.pool.size not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.manager.thread.pool.size Remote log thread pool size, which is used in scheduling tasks to copy segments, fetch remote log indexes and clean up remote log segments."
SR-7 Implement Remote Log Manager Task Interval ConfigurationMatches KAFKA-12816
Add a system-wide configuration property `remote.log.manager.task.interval.ms` that specifies the interval in milliseconds at which the remote log manager runs scheduled tasks such as copying segments to remote storage, fetching remote log indexes, and cleaning up remote log segments. The configuration must accept a positive integer value representing milliseconds.
Acceptance criteria (5)
- A new configuration property `remote.log.manager.task.interval.ms` is added to the broker configuration schema
- The property accepts a positive integer value in milliseconds
- The property is validated at startup to reject zero, negative, or non-integer values
- The remote log manager scheduler uses this interval to schedule topic-partition processing tasks
- After a topic-partition is successfully processed, its next scheduled processing time is set to (current_time + remote.log.manager.task.interval.ms)
Clarification questions (2)
- CLARIFICATION NEEDED: Default value for remote.log.manager.task.interval.ms not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.manager.task.interval.ms The interval at which remote log manager runs the scheduled tasks like copy segments, fetch remote log indexes and clean up remote log segments."
SR-8 Implement Remote Log Reader Thread Pool Size ConfigurationMatches KAFKA-12816
Add a system-wide configuration property `remote.log.reader.threads` that specifies the number of threads in the remote log reader thread pool. This thread pool handles consumer fetch requests that require reading data from remote storage. The configuration must accept a positive integer value.
Acceptance criteria (5)
- A new configuration property `remote.log.reader.threads` is added to the broker configuration schema
- The property accepts a positive integer value
- The property is validated at startup to reject zero, negative, or non-integer values
- The remote storage fetcher thread pool is initialized with the configured number of threads
- Each thread processes one remote fetch request at a time
Clarification questions (2)
- CLARIFICATION NEEDED: Default value for remote.log.reader.threads not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.reader.threads Remote log reader thread pool size"
SR-9 Implement Remote Log Reader Maximum Pending Tasks ConfigurationMatches KAFKA-12816
Add a system-wide configuration property `remote.log.reader.max.pending.tasks` that specifies the maximum size of the remote log reader thread pool task queue. If the task queue is full, the broker must stop accepting new remote log read requests. The configuration must accept a positive integer value.
Acceptance criteria (6)
- A new configuration property `remote.log.reader.max.pending.tasks` is added to the broker configuration schema
- The property accepts a positive integer value
- The property is validated at startup to reject zero, negative, or non-integer values
- The remote storage fetcher thread pool task queue is initialized with the configured maximum capacity
- When the task queue is full, the broker rejects new remote fetch requests or applies appropriate backpressure
- Rejected requests are handled with a defined error response to the client
Clarification questions (3)
- CLARIFICATION NEEDED: Default value for remote.log.reader.max.pending.tasks not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
- CLARIFICATION NEEDED: Exact error response or backpressure mechanism when queue is full not specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Thread pools", statement "remote.log.reader.max.pending.tasks Maximum remote log reader thread pool task queue size. If the task queue is full, broker will stop reading remote log segments."
SR-10 Implement Per-Topic Remote Log Retention Time ConfigurationMatches KAFKA-12816
Add a per-topic configuration property `remote.log.retention.minutes` that specifies the retention period in minutes for remote log segments. This value defines how long log segments are retained in remote storage before being eligible for deletion. The configuration must accept a positive integer value representing minutes.
Acceptance criteria (5)
- A new topic-level configuration property `remote.log.retention.minutes` is added to the topic configuration schema
- The property accepts a positive integer value in minutes
- The property can be set at topic creation or via alter-config operations
- The retention calculation for remote log segments uses this value to determine eligibility for deletion
- If not set, the broker must have a defined default behavior or require the value to be set explicitly when tiered storage is enabled
Clarification questions (3)
- CLARIFICATION NEEDED: Default value for remote.log.retention.minutes not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
- CLARIFICATION NEEDED: Interaction or relationship with local retention (log.retention) not fully specified
Traced to: KIP-405 Section "Public Interfaces > Configs > Per Topic Configuration", statement "remote.log.retention.minutes"
SR-11 Implement Per-Topic Remote Log Retention Bytes ConfigurationMatches KAFKA-12816
Add a per-topic configuration property `remote.log.retention.bytes` that specifies the retention limit in bytes for remote log segments. This value defines the maximum total size of log segments retained in remote storage for a topic-partition before older segments are eligible for deletion. The configuration must accept a positive integer value representing bytes.
Acceptance criteria (5)
- A new topic-level configuration property `remote.log.retention.bytes` is added to the topic configuration schema
- The property accepts a positive integer value in bytes
- The property can be set at topic creation or via alter-config operations
- The retention calculation for remote log segments uses this value to determine eligibility for deletion based on cumulative size
- If not set, the broker must have a defined default behavior or require the value to be set explicitly when tiered storage is enabled
Clarification questions (3)
- CLARIFICATION NEEDED: Default value for remote.log.retention.bytes not specified
- CLARIFICATION NEEDED: Minimum and maximum acceptable values not specified
- CLARIFICATION NEEDED: Interaction or relationship with remote.log.retention.minutes not specified (e.g., OR vs AND logic)
Traced to: KIP-405 Section "Public Interfaces > Configs > Per Topic Configuration", statement "remote.log.retention.bytes"
SR-12 Enforce Compacted Topics Exclusion from Remote Storage
Implement logic to prevent compacted topics from using remote log storage. The system must detect at topic creation, configuration change, or broker startup whether a topic has cleanup.policy=compact and, if remote storage is configured or enabled for that topic, reject the configuration or disable remote storage for that topic.
Acceptance criteria (4)
- During topic creation, if both compaction is enabled (cleanup.policy includes "compact") and remote storage is requested, the request is rejected with a clear error message
- During topic configuration update (alter-config), if compaction is enabled and remote storage is already active, or if remote storage is enabled and compaction is active, the request is rejected with a clear error message
- At broker startup, topics with both compaction and remote storage enabled are logged with a warning and remote storage is disabled for those topics
- The error message explicitly states that compacted topics are not supported with remote storage
Clarification questions (1)
- CLARIFICATION NEEDED: Exact rejection behavior at topic creation vs runtime not fully specified (reject creation vs disable feature)
Traced to: KIP-405 Section "Public Interfaces", statement "Compacted topics will not have remote storage support." and Section "Non-Goals", statement "It does not support compact topics."
SR-13 Define RemoteStorageManager Interface SpecificationMatches KAFKA-9548 KAFKA-9554
Define the RemoteStorageManager interface as specified in the design document. This interface provides the lifecycle operations for remote log segments and indexes, including copying log segments to remote storage, fetching log segment data, and deleting remote log segments. The interface must be the integration point for pluggable remote storage backend implementations.
Acceptance criteria (7)
- A Java interface named RemoteStorageManager is created with all methods specified in the design document
- The interface includes a copyLogSegment method that accepts RemoteLogSegmentId and LogSegmentData and returns success/failure indication
- The interface includes a fetchLogSegmentData method that accepts RemoteLogSegmentMetadata, startPosition (Long), and optional endPosition (Optional<Long>) and returns a Records instance
- The interface includes a deleteLogSegment method that accepts RemoteLogSegmentId
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > Remote Storage Manager", statement "RemoteStorageManager is an interface to provide the lifecycle of remote log segments and indexes." and "[code omitted]"
SR-14 Define RemoteLogMetadataManager Interface SpecificationMatches KAFKA-9548 KAFKA-9554 KAFKA-9555 KAFKA-12429
Define the RemoteLogMetadataManager interface as specified in the design document. This interface provides the lifecycle operations for metadata about remote log segments with strongly consistent semantics. The interface must support storing, fetching, and managing remote log segment metadata, and must allow for pluggable custom implementations while providing a default implementation backed by an internal Kafka topi…
Acceptance criteria (8)
- A Java interface named RemoteLogMetadataManager is created with all methods specified in the design document
- The interface includes a putRemoteLogSegmentData method that accepts RemoteLogSegmentId and RemoteLogSegmentMetadata
- The interface includes methods to fetch remote log segment metadata for a given offset
- The interface includes a method to register a topic partition for metadata cache building
- The interface includes an earliestLogOffset method that accepts TopicPartition and returns the earliest offset tracked in remote storage
- The interface includes a configure method to receive implementation-specific configuration properties
- The interface includes lifecycle methods (e.g., close) as specified
- All method signatures, parameter types, and return types match the design document exactly
Clarification questions (1)
- CLARIFICATION NEEDED: Complete method signatures and return types not provided in the input (marked as "[code omitted]")
Traced to: KIP-405 Section "Public Interfaces > RemoteLogMetadataManager", statement "RemoteLogMetadataManager is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics." and "[code omitted]"
SR-15 Implement RemoteLogManager Component Initialization
Implement the RemoteLogManager (RLM) component that receives callback events for leadership changes and stop/delete events of topic partitions on a broker. RLM must delegate copy, read, and delete operations of topic partition segments to the configured RemoteStorageManager implementation and maintain remote log segment metadata through the RemoteLogMetadataManager. RLM must be initialized only when remote.log.storag…
Acceptance criteria (10)
- A RemoteLogManager class is created as a core broker component
- RLM is initialized during broker startup only if remote.log.storage.enable is true
- RLM receives callbacks from ReplicaManager for topic-partition leadership changes (onPartitionLeadershipChange)
- RLM receives callbacks for topic-partition stop/delete events
- RLM initializes the configured RemoteStorageManager instance using remote.log.storage.manager.class.name
- RLM initializes the configured or default RemoteLogMetadataManager instance
- RLM maintains a registry of managed topic-partitions
- RLM delegates copyLogSegment operations to RemoteStorageManager
- RLM delegates metadata storage operations to RemoteLogMetadataManager
- RLM handles component lifecycle (start, stop, close) correctly
Traced to: KIP-405 Section "Proposed Changes > High-level design", statement "RemoteLogManager (RLM) is a new component which receives callback events for leadership changes and stop/delete events of topic partitions on a broker. delegates copy, read, and delete of topic partition segments to a pluggable storage manager(viz RemoteStorageManager) implementation and maintains respective remote log segment metadata through RemoteLogMetadataManager."
SR-16 Implement RLM Leader Task for Copying Log Segments to Remote StorageMatches KAFKA-9550
Implement the RLM Leader Task that runs for each leader topic partition. This task is responsible for identifying inactive log segment files that meet the criteria for remote upload and copying them to remote storage. The task must process segments one by one from earliest to latest, generate a unique RemoteLogSegmentId for each segment, store metadata via RemoteLogMetadataManager, and invoke copyLogSegment on Remote…
Acceptance criteria (11)
- For each topic-partition where the broker is the leader, an RLM Leader Task is created and scheduled
- The task identifies inactive log segment files (not the active segment)
- The task filters segments where the offset range is not already covered by segments in remote storage
- The task filters segments where the last offset is less than the last-stable-offset of the partition
- Eligible segments are processed in order from earliest to latest
- For each segment, a universally unique RemoteLogSegmentId is generated
- The task calls RemoteLogMetadataManager.putRemoteLogSegmentData with the RemoteLogSegmentId and RemoteLogSegmentMetadata
- The task calls RemoteStorageManager.copyLogSegment with the RemoteLogSegmentId and LogSegmentData
- If copyLogSegment is successful, the metadata is updated with success status
- If copyLogSegment fails, the metadata entry is removed
- The task handles failures gracefully and schedules retry according to the configured retry interval
Traced to: KIP-405 Section "Proposed Changes > High-level design" and "RLM/RSM tasks and thread pools > 1. Remote Log Manager (RLM) Thread Pool > Copy log segments to remote storage (leader)", full paragraph starting with "Copy the log segment files that are inactive..."
SR-17 Implement RLM Task for Handling Expired Remote Segments (Leader and Follower)Matches KAFKA-14888
Implement the RLM task that handles expiration and deletion of remote log segments based on remote retention configuration. The leader computes segments to be deleted, updates the earliest offset in RemoteLogMetadataManager, removes segment metadata and data from remote storage, and handles failures by logging them to a dedicated topic. Followers fetch the earliest offset and update their local indexes and start offs…
Acceptance criteria (9)
- For leader replicas, the task computes expired remote log segments based on remote.log.retention.minutes and remote.log.retention.bytes
- The task updates the earliest offset for the topic partition in RemoteLogMetadataManager
- The task retrieves all remote log segment IDs eligible for deletion
- The task calls RemoteStorageManager.deleteLogSegment for each segment ID
- The task removes metadata entries via RemoteLogMetadataManager
- If deletion from remote storage fails, the failure is logged with the RemoteLogSegmentId to a dedicated topic (default: __remote_segments_to_be_deleted)
- For follower replicas, the task calls RemoteLogMetadataManager.earliestLogOffset(TopicPartition) to fetch the earliest offset
- Followers clean up local indexes up to the received earliest offset
- Followers update their local log start offset with the received value
Clarification questions (2)
- CLARIFICATION NEEDED: Exact name and schema of the __remote_segments_to_be_deleted topic not fully specified
- CLARIFICATION NEEDED: Cleanup mechanism for entries in __remote_segments_to_be_deleted topic not specified
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 1. Remote Log Manager (RLM) Thread Pool > Handle expired remote segments (leader and follower)", full paragraph starting with "RLM leader computes the log segments to be deleted..."
SR-18 Implement RLM Topic-Partition Scheduled Processing Logic
Implement the scheduling logic where RemoteLogManager maintains a list of managed topic-partitions, each with a scheduled processing time. Topic-partitions are processed by the RLM thread pool when their scheduled time is less than or equal to the current time. After successful processing, the scheduled time is updated to (now + rlm_process_interval_ms). After failure, the scheduled time is updated to (now + rlm_retr…
Acceptance criteria (8)
- RLM maintains a thread-safe data structure (list or priority queue) of managed topic-partitions with scheduled processing times
- The list is updated in Kafka I/O threads when topic-partitions are added or removed
- When a new topic-partition is assigned, it is added to the list with scheduled processing time = 0 (immediate processing)
- The RLM thread pool continuously selects topic-partitions where scheduled_time <= current_time for processing
- After a topic-partition is successfully processed, its scheduled time is set to (current_time + remote.log.manager.task.interval.ms)
- After a topic-partition processing fails due to remote storage error, its scheduled time is set to (current_time + rlm_retry_interval_ms)
- When a topic-partition is unassigned and not currently being processed, it is immediately removed from the list
- When a topic-partition is unassigned and currently being processed, it is marked as "deleted" and removed after the current processing completes
Clarification questions (1)
- CLARIFICATION NEEDED: The configuration property name "rlm_retry_interval_ms" is not defined in the Configs section; only "remote.log.manager.task.interval.ms" is defined. Are these the same or is rlm_retry_interval_ms a separate config?
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 1. Remote Log Manager (RLM) Thread Pool", paragraph starting with "RLM maintains a list of the topic-partitions it manages..."
SR-19 Implement Remote Storage Fetcher Thread Pool and RemoteFetchPurgatoryMatches KAFKA-9579 KAFKA-14889
Implement the remote storage fetcher thread pool that processes consumer fetch requests requiring data from remote storage. When a fetch request targets an offset in remote storage, the request is added to a RemoteFetchPurgatory (an instance of DelayedOperationPurgatory) to handle timeout, and the request is queued in the remote storage fetcher thread pool task queue. Each thread processes one remote fetch request by…
Acceptance criteria (12)
- A RemoteFetchPurgatory instance is created as a DelayedOperationPurgatory for remote fetch requests
- A remote storage fetcher thread pool is created with size configured by remote.log.reader.threads
- The thread pool task queue has maximum capacity configured by remote.log.reader.max.pending.tasks
- When a fetch request targets an offset in remote storage, it is added to RemoteFetchPurgatory and queued in the thread pool
- Each thread processes one remote fetch request at a time
- The thread finds the RemoteLogSegmentId for the requested offset from RemoteLogMetadataManager
- The thread determines startPosition and endPosition from the offset index
- The thread calls RemoteStorageManager.fetchLogSegmentData(RemoteLogSegmentMetadata, startPosition, endPosition) to retrieve data
- The thread builds a Records instance from the fetched data
- If the remote segment file is deleted, RemoteFetchPurgatory is notified to return an error to the client
- If the remote storage operation fails, the operation is retried with Exponential Back-Off until the original consumer fetch request timeout
- The thread pool enforces the maximum pending tasks limit and rejects or applies backpressure when the queue is full
Clarification questions (2)
- CLARIFICATION NEEDED: Exact Exponential Back-Off parameters (initial delay, multiplier, max delay) not specified
- CLARIFICATION NEEDED: Error response type when remote segment is deleted not specified
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 2. Remote Storage Fetcher Thread Pool", full paragraph starting with "When handling consumer fetch request..."
SR-20 Implement Local and Remote Log Offset Constraints Enforcement
Implement logic to maintain and enforce the offset constraints for leader topic partitions with tiered storage enabled. The constraints define the relationships between local log start offset (Lx), local log end offset (Lz), last stable offset (Ly), remote log end offset (Ry), and remote log start offset (Rx). These constraints must be validated and maintained during segment upload, deletion, and offset updates.
Acceptance criteria (8)
- The system maintains the following offset values for each leader topic partition: Lx (local log start offset), Lz (local log end offset), Ly (last stable offset), Ry (remote log end offset), Rx (remote log start offset)
- The system enforces the constraint: Lz >= Ly
- The system enforces the constraint: Ly >= Lx
- The system enforces the constraint: Ly >= Ry
- The system enforces the constraint: Ry >= Rx
- These constraints are validated before and after segment copy operations
- These constraints are validated before and after remote segment deletion operations
- Any operation that would violate a constraint is rejected or deferred until the constraint can be satisfied
Traced to: KIP-405 Section "Proposed Changes > Local and Remote log offset constraints", full paragraph listing offset variables and constraints
SR-21 Implement Follower Local Segment Retention Logic for Tiered Storage
Implement logic for follower replicas to defer deletion of local segment files until the corresponding segment data has been retrieved from remote storage and local indexes have been updated. Local segment files can only be deleted when: (1) the follower has received the corresponding segment data info from remote storage and updated its index files, AND (2) the local files are older than the configured remote retent…
Acceptance criteria (6)
- For follower replicas, segment deletion checks whether the segment metadata is present in RemoteLogMetadataManager
- For follower replicas, segment deletion checks whether the segment has exceeded the remote retention time
- A segment file is deleted only if both conditions are satisfied
- If the segment metadata is not yet available, deletion is deferred even if the local retention time has expired
- Follower replicas consume remote log metadata topic partitions to receive segment metadata updates
- Upon receiving metadata, followers update their local remote log indexes
Traced to: KIP-405 Section "Proposed Changes > Manage Remote Log Segments", paragraph starting with "The leader may fail to ship segment data to remote storage on time..."
SR-22 Implement ReplicaManager Integration for RLM Topic-Partition Lifecycle
Integrate RemoteLogManager with ReplicaManager to handle topic-partition assignment and removal events. When tiered storage is configured, ReplicaManager must call RLM to assign or remove topic-partitions. When a broker transitions from Leader to Follower, RLM must complete any in-progress segment copy operation before relinquishing control, which may result in duplicate segments that are cleaned up later based on re…
Acceptance criteria (6)
- ReplicaManager checks if RemoteLogManager is configured (remote.log.storage.enable is true)
- When a topic-partition is assigned to the broker, ReplicaManager calls RemoteLogManager to register the topic-partition
- When a topic-partition is removed from the broker, ReplicaManager calls RemoteLogManager to unregister the topic-partition
- When a broker transitions from Leader to Follower for a topic-partition, RemoteLogManager is notified
- If a segment copy operation is in progress during Leader to Follower transition, the copy completes before the transition is finalized
- Duplicate segments resulting from the transition are handled during subsequent remote retention cleanup
Traced to: KIP-405 Section "Proposed Changes > Replica Manager", paragraph starting with "If RLM is configured, ReplicaManager will call RLM..."
SR-23 Implement Consumer Fetch Request Delegation to RemoteLogManagerMatches KAFKA-14889
Modify consumer fetch request handling in ReplicaManager to delegate reads to RemoteLogManager when the requested offset is not in local storage. For any fetch request, ReplicaManager calls readFromLocalLog; if this method returns an OffsetOutOfRangeException, ReplicaManager delegates the read call to RemoteLogManager to serve the data from remote storage.
Acceptance criteria (6)
- ReplicaManager fetch request handling first attempts to read from local log via readFromLocalLog
- If readFromLocalLog succeeds, the data is returned to the client as normal
- If readFromLocalLog throws OffsetOutOfRangeException, ReplicaManager checks if the offset is within the remote log offset range
- If the offset is in the remote log range, ReplicaManager delegates the read to RemoteLogManager
- RemoteLogManager handles the remote fetch as per the remote storage fetcher thread pool logic
- If the offset is not in the remote log range (i.e., before remote log start offset or after log end offset), an appropriate error is returned to the client
Traced to: KIP-405 Section "Proposed Changes > Consumer Fetch Requests", paragraph starting with "For any fetch requests, ReplicaManager will proceed..."
SR-24 Implement Follower Replication Fetch Request Handling for Tiered StorageMatches KAFKA-13369
Modify follower fetch request handling so that leaders only return data that is still in the leader's local storage. Log segments that exist only in remote storage are not replicated to followers, as they are already present in remote storage. Followers retrieve information about remote segments from RemoteLogMetadataManager instead of replicating the data.
Acceptance criteria (5)
- For follower fetch requests, the leader identifies the requested offset range
- The leader serves data only from local log segments
- If the requested offset range includes offsets that exist only in remote storage, those offsets are not included in the fetch response
- Followers are expected to retrieve remote segment metadata from RemoteLogMetadataManager to learn about remote segments
- If a follower becomes a leader, it can locate and serve data from remote storage using the metadata it has retrieved
Clarification questions (1)
- CLARIFICATION NEEDED: Exact mechanism for followers to know they should not request remote-only offsets not fully specified (is it implicit in the offset range or explicit signaling?)
Traced to: KIP-405 Section "Proposed Changes > Follower Requests/Replication", paragraph starting with "For follower fetch, the leader only returns the data..."
SR-25 Implement DeleteRecords API Semantics for Tiered Storage
Ensure the DeleteRecords API semantics are unchanged when tiered storage is enabled. The API deletes records until the given offset if possible, which is equivalent to updating logStartOffset of the partition log. If the new logStartOffset extends into remote storage, the system must asynchronously clean remote logs after updating the log-start-offset.
Acceptance criteria (5)
- DeleteRecords API accepts an offset value
- The API updates the logStartOffset of the partition log with the given offset if it is greater than the current log-start-offset and less than or equal to high-watermark
- The API operation succeeds and returns to the client without waiting for remote log cleanup
- After updating log-start-offset, if the new offset extends beyond the previous remote log start offset, the system schedules an asynchronous task to clean remote logs
- Remote log cleanup updates the earliest offset in RemoteLogMetadataManager and deletes expired segments from RemoteStorageManager
Traced to: KIP-405 Section "Proposed Changes > Other APIs > DeleteRecords", paragraph starting with "There is no change in the semantics of this API..."
SR-26 Implement ListOffsets API Enhancement for Tiered Storage
Enhance the ListOffsets API to support timestamps >= 0 by checking both remote log time indexes and local log time indexes. For timestamps >= 0, return the first message offset whose timestamp is >= the given timestamp, checking remote log time indexes first, then local log time indexes. The API must also support existing timestamp values: EARLIEST_TIMESTAMP (-2) returns logStartOffset, LATEST_TIMESTAMP (-1) returns …
Acceptance criteria (8)
- ListOffsets API accepts a timestamp value
- For timestamp = -2 (EARLIEST_TIMESTAMP), return the logStartOffset of the log (local or remote, whichever is earliest)
- For timestamp = -1 (LATEST_TIMESTAMP), return log-stable-offset or log-end-offset based on the isolation level in the request
- For timestamp >= 0, first search remote log time indexes for the first offset with timestamp >= requested timestamp
- If found in remote log, return that offset
- If not found in remote log, search local log time indexes for the first offset with timestamp >= requested timestamp
- If found in local log, return that offset
- If not found in either, return an appropriate error or end-of-log indication
Traced to: KIP-405 Section "Proposed Changes > Other APIs > ListOffsets", paragraph starting with "ListOffsets API gives the offset(s) for the given timestamp..." up to "For timestamps >= 0, it returns the first message offset..."
SR-27 Implement ListOffsets API New Target Timestamp NEXT_LOCAL_TIMESTAMPMatches KAFKA-13369
Add a new target timestamp value NEXT_LOCAL_TIMESTAMP (value -3) to the ListOffsets API. This value is used by follower replicas to request the offset that they should start fetching to replicate local logs, where all records earlier than this offset can be considered as copied to remote storage. The request and response schemas do not add new fields, but the version must be bumped to indicate support for the new tim…
Acceptance criteria (6)
- ListOffsets API supports a new target timestamp value: NEXT_LOCAL_TIMESTAMP with value -3
- The ListOffsets request and response schemas do not add new fields
- The ListOffsets request and response version is incremented to indicate support for NEXT_LOCAL_TIMESTAMP
- When a follower sends a request with target timestamp = -3 (NEXT_LOCAL_TIMESTAMP), the leader returns the first offset in the local log (i.e., the earliest offset not yet copied to remote storage or the first offset that followers must replicate)
- The returned offset allows followers to avoid fetching records that are already in remote storage
- The implementation documents the use case: followers use this to determine the starting offset for replication
Clarification questions (1)
- CLARIFICATION NEEDED: Exact offset calculation for NEXT_LOCAL_TIMESTAMP not fully specified (is it remote_log_end_offset + 1, or local_log_start_offset, or something else?)
Traced to: KIP-405 Section "Proposed Changes > Other APIs > ListOffsets", paragraph starting with "This API is enhanced with supporting new target timestamp value as -3..."
SR-28 Implement JBOD Alter Log Dirs Handling for Tiered Storage
Ensure that JBOD (Just a Bunch Of Disks) alter log dirs operations are compatible with tiered storage. JBOD supports altering log directories to a different directory on the local broker or to a directory on a remote broker. The tiered storage implementation must not interfere with these operations.
Acceptance criteria (4)
- Alter log dirs to a different directory on the local broker continues to work when tiered storage is enabled
- Alter log dirs to a directory on a remote broker continues to work when tiered storage is enabled
- Remote log metadata and indexes are updated or moved as necessary when log directories are altered
- The remote storage backend is not affected by local log directory changes
Clarification questions (1)
- CLARIFICATION NEEDED: Detailed interaction between JBOD alter log dirs and remote log metadata/indexes not fully specified
Traced to: KIP-405 Section "Proposed Changes > Other APIs > JBOD related changes", paragraph starting with "Currently, JBOD is supported by altering log dirs..."
SR-29 Define Default Remote Log Metadata Topic Name and Configuration
Define the default name for the internal remote log metadata topic as `__remote_log_metadata`. This topic stores metadata about remote log segments and is consumed by RemoteLogMetadataManager. Users must be able to configure the topic name, the number of partitions (default 50), and the replication factor.
Acceptance criteria (7)
- The default remote log metadata topic name is `__remote_log_metadata`
- The default number of partitions is 50
- Users can configure the topic name via a configuration property
- Users can configure the number of partitions via a configuration property
- Users can configure the replication factor via a configuration property
- The topic is created automatically by the broker at startup if it does not exist and tiered storage is enabled
- The topic creation respects the configured name, partitions, and replication factor
Clarification questions (2)
- CLARIFICATION NEEDED: Exact configuration property names for topic name, partition count, and replication factor not specified
- CLARIFICATION NEEDED: Default replication factor not specified
Traced to: KIP-405 Section "Proposed Changes > RemoteLogMetadataManager implemented with an internal topic", paragraph starting with "Metadata of remote log segments are stored in an internal topic called `__remote_log_metadata`..."
SR-30 Implement Remote Log Metadata Topic Partition Assignment Logic
Implement the logic to determine which remote log metadata topic partition is used for a given user topic-partition. The partition is computed as: user-topic-partition.toString().hashCode() % no_of_remote_log_metadata_topic_partitions. This ensures consistent assignment of metadata for a user topic-partition to a specific metadata topic partition.
Acceptance criteria (5)
- For each user topic-partition, the remote log metadata topic partition is computed using the formula: user-topic-partition.toString().hashCode() % number_of_remote_log_metadata_topic_partitions
- The hashCode() method is called on the string representation of the TopicPartition object
- The modulo operation uses the configured number of remote log metadata topic partitions
- The computed partition is used consistently for all metadata operations for that user topic-partition
- The logic is documented and unit-tested with known inputs and outputs
Traced to: KIP-405 Section "Proposed Changes > RemoteLogMetadataManager implemented with an internal topic", statement "Remote log metadata topic partition for a given user topic-partition is: user-topic-partition.toString().hashCode() % no_of_remote_log_metadata_topic_partitions"
SR-31 Implement RemoteLogMetadataManager Cache for Leader and Follower ReplicasMatches KAFKA-12802
Implement cache management logic in RemoteLogMetadataManager where leader replicas update the cache by producing metadata to the remote log metadata topic, and follower replicas build and maintain the cache by consuming from the respective remote log metadata topic partitions. The cache must be disk-backed (e.g., RocksDB) to handle potentially large metadata sets without excessive memory usage.
Acceptance criteria (7)
- RemoteLogMetadataManager maintains an in-memory or disk-backed cache of remote log segment metadata
- For leader replicas, after copying a segment to remote storage, RemoteLogManager calls RemoteLogMetadataManager.putRemoteLogSegmentData, which writes the metadata to the remote log metadata topic and updates the local cache
- For follower replicas, RemoteLogMetadataManager subscribes to the relevant remote log metadata topic partitions
- For follower replicas, RemoteLogMetadataManager consumes metadata messages and adds them to the local cache
- When a topic partition is reassigned to a new broker, RemoteLogMetadataManager subscribes to the corresponding remote log metadata topic partition if not already subscribed
- The cache is backed by a persistent store (e.g., RocksDB) to minimize memory footprint
- Committed offsets for consumed metadata topic partitions are stored in a local file to avoid re-reading messages after broker restart
Clarification questions (2)
- CLARIFICATION NEEDED: Exact disk-backed cache implementation (RocksDB confirmed as example, but not mandated)
- CLARIFICATION NEEDED: Local file format and location for storing committed offsets not specified
Traced to: KIP-405 Section "Proposed Changes > RemoteLogMetadataManager implemented with an internal topic", paragraphs starting with "RLMM registers the topic partitions..." and "This requires the cache to be based on disk storage like RocksDB..."
SR-32 Define Remote Log Index File Formats (remoteLogIndex, remoteOffsetIndex, remoteTimeIndex)
Define the file formats for three remote log index file types: remoteLogIndex, remoteOffsetIndex, and remoteTimeIndex. The remoteLogIndex file contains entries mapping offset ranges to Remote Data Identifiers (RDI) in remote storage. The remoteOffsetIndex and remoteTimeIndex files are similar to existing Kafka .index and .timeindex files but point to entries in the remoteLogIndex file instead of log segment files. In…
Acceptance criteria (6)
- The remoteLogIndex file format is defined with the following fields per entry: magic (int16, current value 0), length (int16), crc (int32), firstOffset (int64), lastOffset (int64), firstTimestamp (int64), lastTimestamp (int64), dataLength (int32), rdiLength (int16), rdi (byte[])
- The remoteOffsetIndex file format is defined, with entries pointing to remoteLogIndex entries instead of log segment positions
- The remoteTimeIndex file format is defined, with entries pointing to remoteLogIndex entries instead of log segment positions
- Index files are named with the base offset of the first remote segment they index: {base_offset}.remoteLogIndex, {base_offset}.remoteOffsetIndex, {base_offset}.remoteTimeIndex
- Index files are rotated at a configurable time interval or size threshold
- RDI (Remote Data Identifier) format is implementation-dependent and encodes the location (e.g., HDFS path, S3 key) and offset of the data in remote storage
Clarification questions (2)
- CLARIFICATION NEEDED: Configurable time interval and size threshold for index file rotation not specified
- CLARIFICATION NEEDED: Exact RDI format for different remote storage implementations not specified (noted as implementation-dependent)
Traced to: KIP-405 Section "Proposed Changes > Remote Log Indexes", paragraph starting with "For each topic partition that has RLM configured..." and format specification starting with "magic: int16..."
SR-33 Implement Remote Log Index File Generation and Upload for Leader Replicas
Implement the logic for leader replicas to generate remoteLogIndex, remoteOffsetIndex, and remoteTimeIndex entries after copying a log segment to remote storage. The remoteLogIndex entries must be appended locally and also shipped to remote storage along with the segment data. Followers retrieve these index entries from remote storage to build their own indices.
Acceptance criteria (7)
- After a log segment is copied to remote storage, the leader generates one or more remoteLogIndex entries for that segment
- Each remoteLogIndex entry contains the offset range, timestamp range, data length, and RDI (Remote Data Identifier) for a portion of the remote segment
- The remoteLogIndex entries are appended to the local remoteLogIndex file
- Corresponding remoteOffsetIndex and remoteTimeIndex entries are generated and appended to the respective local index files
- The remoteLogIndex entries are uploaded to remote storage along with the segment data
- Followers retrieve the remoteLogIndex entries from remote storage
- Followers use the retrieved entries to build their own local remoteLogIndex, remoteOffsetIndex, and remoteTimeIndex files
Clarification questions (1)
- CLARIFICATION NEEDED: Number of remoteLogIndex entries per segment (one or more) and granularity decision criteria not specified
Traced to: KIP-405 Section "Proposed Changes > Remote Log Indexes", paragraph starting with "Depends on the implementation, RLM may append 1 or more entries..." and "The RemoteLogIndex entries are shipped to remote storage along with the segment data..."
SR-34 Implement Bounded LRU Cache for Remote Log Index Files
Implement a bounded LRU (Least Recently Used) cache in RemoteLogManager to store remote log index files (remoteLogIndex, remoteOffsetIndex, remoteTimeIndex) fetched from remote storage. This cache avoids repeated fetches of the same index files and allows them to be used in the same way as local segment indexes for data retrieval.
Acceptance criteria (7)
- RemoteLogManager maintains an LRU cache for remote log index files
- The cache is bounded by a configurable maximum size (number of index files or total bytes)
- When an index file is needed for a remote fetch, RemoteLogManager checks the cache first
- If the index file is in the cache, it is used directly without fetching from remote storage
- If the index file is not in the cache, it is fetched from remote storage and added to the cache
- The LRU eviction policy removes the least recently used index file when the cache is full
- The cache is thread-safe and supports concurrent access by multiple remote fetch threads
Clarification questions (1)
- CLARIFICATION NEEDED: Configurable maximum cache size (number of files or bytes) not specified
Traced to: KIP-405 Section "Proposed Changes > High-level design", statement "RLM maintains a bounded cache(possibly LRU) of the index files of remote log segments to avoid multiple index fetches from the remote storage."
SR-35 Define Default Topic Name for Failed Remote Segment Deletion Tracking
Define the default name for the internal topic used to track remote log segments that failed to be deleted from remote storage. The default topic name is `__remote_segments_to_be_deleted`. This topic stores RemoteLogSegmentId values for segments that could not be removed from remote storage during the cleanup process, allowing users or automated processes to consume these events and retry deletion.
Acceptance criteria (6)
- The default topic name for failed deletion tracking is `__remote_segments_to_be_deleted`
- Users can configure the topic name via a configuration property
- When deletion of a remote log segment via RemoteStorageManager fails, the RemoteLogSegmentId is written to this topic
- The topic schema includes at minimum the RemoteLogSegmentId
- Users or automated processes can consume events from this topic and retry deletion manually or programmatically
- The topic is created automatically by the broker if it does not exist and tiered storage is enabled
Clarification questions (3)
- CLARIFICATION NEEDED: Configuration property name for topic name not specified
- CLARIFICATION NEEDED: Complete topic schema (key, value format) not specified
- CLARIFICATION NEEDED: Automated retry mechanism for failed deletions not specified
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools > 1. Remote Log Manager (RLM) Thread Pool > Handle expired remote segments", statement "If there are any failures in removing remote log segments then those are stored in a specific topic (default as __remote_segments_to_be_deleted)..."
SR-36 Implement Remote Storage Availability Isolation to Prevent Impact on Core Kafka Operations
Implement architectural isolation to ensure that when remote storage becomes temporarily unavailable (up to several hours) or experiences high latency (up to minutes), core Kafka operations that do not rely on remote storage continue to function normally. All operations such as produce, consume local data, create/expand topics, and replica management must not be impacted by remote storage unavailability. Consumers at…
Acceptance criteria (9)
- All remote storage operations (copy, fetch, delete) are executed in dedicated thread pools separate from Kafka I/O threads and fetcher threads
- Kafka I/O threads and fetcher threads do not block or wait on remote storage operations
- When remote storage is unavailable, produce operations to local log continue without delay
- When remote storage is unavailable, consumer fetch operations for local data continue without delay
- When remote storage is unavailable, topic creation and expansion operations continue without delay
- When remote storage is unavailable, replica management and leader election operations continue without delay
- Consumers attempting to fetch data from remote storage during unavailability receive a clear error response (e.g., timeout or unavailable error)
- Remote storage operation failures are logged with appropriate severity and detail
- The system retries failed remote storage operations according to the configured retry interval without blocking other operations
Traced to: KIP-405 Section "RLM/RSM tasks and thread pools", paragraph starting with "Remote storage (e.g. HDFS/S3/GCP) is likely to have higher I/O latency..." and "When the remote storage becoming temporarily unavailable... Kafka should still be able to operate normally..."
SR-37 Implement Goal: Extend Kafka Storage Beyond Local Disks with Minimal Internal Impact
Implement the core goal of tiered storage: extend Kafka's storage beyond the local storage available on the Kafka cluster by retaining older data in an external store (such as HDFS or S3) with minimal impact on the internals of Kafka. Kafka behavior and operational complexity must not change for existing users that do not have tiered storage feature configured.
Acceptance criteria (7)
- Older log segments are copied to remote storage (HDFS, S3, or other implementations of RemoteStorageManager)
- The tiered storage feature is disabled by default (remote.log.storage.enable = false)
- When tiered storage is disabled, Kafka behavior is identical to existing Kafka behavior with no performance or operational impact
- When tiered storage is enabled, local storage requirements are significantly reduced compared to storing all data locally
- The total retention period for data can be extended beyond local disk capacity by using remote storage
- The implementation does not require changes to core Kafka protocols or data formats visible to clients
- Existing producers and consumers operate without modification when tiered storage is enabled
Traced to: KIP-405 Section "Goals", paragraph starting with "Extend Kafka's storage beyond the local storage available on the Kafka cluster..."
SR-38 Implement Non-Goal: Tiered Storage Does Not Replace ETL Pipelines
Document and implement the non-goal that tiered storage does not replace ETL pipelines and jobs. Existing ETL pipelines must continue to consume data from Kafka as usual, with the benefit that data in Kafka now has a much longer retention period due to tiered storage.
Acceptance criteria (4)
- Tiered storage implementation does not alter or replace any existing ETL pipeline functionality
- ETL pipelines can continue to consume data from Kafka using existing consumer APIs
- The extended retention period provided by tiered storage is available to ETL consumers
- Documentation clearly states that tiered storage is not a replacement for ETL and that existing ETL workflows are unaffected
Traced to: KIP-405 Section "Non-Goals", statement "Tiered storage does not replace ETL pipelines and jobs. Existing ETL pipelines continue to consume data from Kafka as is, albeit with data in Kafka having a much longer retention period."
Download the pack
What this does not show
This is one run of a language model on one document. The comparisons were made by strict AI checks and every match we count was read by hand; where a match didn't hold up, we say so (the S3, HDFS and Azure plug-ins, and one report closed as Won't Fix). Kafka isn't regulated medical software, so this shows the method on real-world material, not compliance with any standard. The savings use the published assumptions above and count writing time only; your team still reviews every ticket. SprintReady drafts tickets and flags problems; it doesn't replace your design review or your quality system.
How we ran it
- SprintReady's production ticket-writing instructions and model (eu.anthropic.claude-sonnet-4-5-20250929-v1:0), run once, with no client context and nothing tuned for this test.
- People's names were replaced with labels and code blocks were left out. Otherwise the document is the public text.
- The team's tickets are the 65 sub-tasks, improvements and bugs opened under or linked to KAFKA-7739 from 13 February 2020 onwards.
Source, licence and no endorsement
The design document and tickets are published by the Apache Kafka project. Content contributed to Apache's issue tracker and wiki is licensed under the Apache License 2.0; we've replaced names with labels and left out code blocks. SprintReady is not affiliated with, sponsored by or endorsed by the Apache Software Foundation or the Apache Kafka project. We use the project's public records only as an example of a real specification, because they are public, licensed for reuse and written by engineers with no connection to us. Apache, Apache Kafka and Kafka are trademarks of the Apache Software Foundation, named here only to identify the source. We don't use their logos. Full terms for this example: section 16 of our terms and conditions.
Try it on your own specification: 30 free credits