Status

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA:

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

KIP-405 introduced the Tiered Storage feature, which allows to scale the cluster storage independently from CPU and memory, also improving maintenance operations. For each leader's partition log, there is a periodic RLMCopyTask that copies inactive and locally expired segments to the configured remote storage system, along with their auxiliary state (indexes, producer state snapshots, leader-epoch checkpoints). Each segment copy is associated with a metadata object that contains a UUID for that specific copy operation. By default, these metadata are store in an internal non-compacted topic called __remote_log_metadata, and include the remote segment state, which is initialized to COPY_SEGMENT_STARTED.

When a segment copy fails with a RemoteStorageException, we immediately try to delete the remote segment data, and then a new copy operation is triggered by the next RLMCopyTask execution. If there is a persistent network or remote storage issue, both copy and delete operations fail. The problem is that, while a new copy operation is triggered by the next RLMCopyTask execution and eventually succeed, the delete operation is never retried once it fails. This leaves a certain amount of invalid segment data in the remote storage (wasted space). The consequence of this could be remote storage exhaustion, or unnecessary cloud storage expense. Additionally, the current logic is fragile, because the COPY_SEGMENT_STARTED actually represents more than one segment state (copy is ongoing and copy failed).

Public Interfaces

The RemoteLogSegmentState public enumeration is declared evolving, and it will be changed to add the DANGLING state.

@InterfaceStability.Evolving
public enum RemoteLogSegmentState {

    /**
     * This state indicates that the segment copy and/or deletion failed possibly leaving some data in the remote storage.
     */
    DANGLING((byte) 4);

Proposed Changes

We propose to introduce a new non-terminal remote segment state called DANGLING, which would be set when the segment copy fail (no deletion will be attempted at this stage). After copying all eligible segments, the RLMCopyTask will try to delete each dangling segment, moving the state to DELETE_SEGMENT_STARTED. If the delete operation fails, the state will move back to DANGLING, so that we can retry in the next RLMCopyTask execution, otherwise the remote segment will move to DELETE_SEGMENT_FINISHED, which is a terminal state.

+--------------------+         +---------------------+
|COPY_SEGMENT_STARTED|-------->|COPY_SEGMENT_FINISHED|
+--+--------------+--+         +--+------------------+
   |              |               |
   |              |               |
   v              v               v
+--+-----+     +--+---------------+---+
|DANGLING|---->|DELETE_SEGMENT_STARTED|
+---+----+     +--+--------+----------+
    ^             |        |
    |_____________|        |
                           v
               +-----------+-----------+
               |DELETE_SEGMENT_FINISHED|
               +-----------------------+

New broker metrics will be added to track the current number and total size of dangling remote segments.

Metric/attribute name	Description	MBean name
Dangling Remote Log Count	The total number of dangling segments in remote storage. Omitting 'topic=(...)' will yield the all-topic count.	`kafka.server:type=BrokerTopicMetrics,name=DanglingRemoteLogCount,topic=([-.\w]+)`
Dangling Remote Log Bytes	Bytes which belongs to dangling remote segments, but are not yet deleted. Omitting 'topic=(...)' will yield the all-topic sum.	`kafka.server:type=BrokerTopicMetrics,name=DanglingRemoteLogBytes,topic=([-.\w]+)`

Compatibility, Deprecation, and Migration Plan

There is no backwards compatibility issue.

Test Plan

Existing unit tests will be updated as needed, and a new unit test created to verify the new behavior.

Rejected Alternatives

There is no rejected alternative.