crabka-remote-storage 0.3.6

KIP-405 tiered-storage SPI (RemoteStorageManager / RemoteLogMetadataManager) and reference implementations for Crabka
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

//! The [`RemoteLogMetadataManager`] SPI: persistence and querying of
//! remote-segment metadata.

use crate::error::RemoteStorageError;
use crate::metadata::{
    RemoteLogSegmentMetadata, RemoteLogSegmentMetadataUpdate, RemotePartitionDeleteMetadata,
    TopicIdPartition,
};

/// SPI for the store that holds metadata about remote log segments.
///
/// The reference implementation is
/// [`InmemoryRemoteLogMetadataManager`](crate::InmemoryRemoteLogMetadataManager);
/// a production implementation backs the metadata with an internal topic
/// (Kafka's `TopicBasedRemoteLogMetadataManager`). Implementations must be
/// `Send + Sync`.
///
/// The lifecycle invariants every implementation enforces:
///
/// - A segment is introduced with
///   [`add_remote_log_segment_metadata`](RemoteLogMetadataManager::add_remote_log_segment_metadata)
///   in state
///   [`CopySegmentStarted`](crate::RemoteLogSegmentState::CopySegmentStarted).
/// - Every later change is an
///   [`update_remote_log_segment_metadata`](RemoteLogMetadataManager::update_remote_log_segment_metadata)
///   that follows a valid state transition.
/// - Only
///   [`CopySegmentFinished`](crate::RemoteLogSegmentState::CopySegmentFinished)
///   segments are visible to the offset/epoch queries.
pub trait RemoteLogMetadataManager: Send + Sync {
    /// Record a newly-started segment copy. The metadata's state must be
    /// [`CopySegmentStarted`](crate::RemoteLogSegmentState::CopySegmentStarted)
    /// and its id must not already be known.
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError::InvalidAdd`] if the starting state is
    /// wrong or the segment id is a duplicate.
    fn add_remote_log_segment_metadata(
        &self,
        metadata: RemoteLogSegmentMetadata,
    ) -> Result<(), RemoteStorageError>;

    /// Advance a known segment's lifecycle state.
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError::SegmentNotFound`] if the segment is
    /// unknown, or [`RemoteStorageError::InvalidSegmentTransition`] if the
    /// transition is not permitted.
    fn update_remote_log_segment_metadata(
        &self,
        update: RemoteLogSegmentMetadataUpdate,
    ) -> Result<(), RemoteStorageError>;

    /// Find the finished segment that contains `offset` for the given
    /// `leader_epoch`, if any.
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError`] only on an underlying store failure;
    /// a simple "no such segment" is `Ok(None)`.
    fn remote_log_segment_metadata(
        &self,
        topic_id_partition: &TopicIdPartition,
        leader_epoch: i32,
        offset: i64,
    ) -> Result<Option<RemoteLogSegmentMetadata>, RemoteStorageError>;

    /// The highest offset that the remote tier holds for `leader_epoch`
    /// (the max `end_offset` across finished segments carrying that epoch).
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError`] only on an underlying store failure.
    fn highest_offset_for_epoch(
        &self,
        topic_id_partition: &TopicIdPartition,
        leader_epoch: i32,
    ) -> Result<Option<i64>, RemoteStorageError>;

    /// All segments known for a partition, ordered by `start_offset`
    /// (regardless of state).
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError`] only on an underlying store failure.
    fn list_remote_log_segments(
        &self,
        topic_id_partition: &TopicIdPartition,
    ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError>;

    /// All segments carrying `leader_epoch`, ordered by `start_offset`
    /// (regardless of state).
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError`] only on an underlying store failure.
    fn list_remote_log_segments_by_epoch(
        &self,
        topic_id_partition: &TopicIdPartition,
        leader_epoch: i32,
    ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError>;

    /// Record a partition-delete lifecycle event.
    ///
    /// # Errors
    ///
    /// Returns [`RemoteStorageError::InvalidPartitionDeleteTransition`] if
    /// the transition is not permitted from the partition's current
    /// delete-state.
    fn put_remote_partition_delete_metadata(
        &self,
        metadata: RemotePartitionDeleteMetadata,
    ) -> Result<(), RemoteStorageError>;
}