Struct TopicBasedRemoteLogMetadataManager 

pub struct TopicBasedRemoteLogMetadataManager { /* private fields */ }

Production RemoteLogMetadataManager backed by the __remote_log_metadata topic (via a MetadataEventLog adapter).

Construct with Self::start; it loads any on-disk snapshot but consumes no metadata partitions until Self::reconcile_assignment adds the broker’s leader/follower-derived set.

Implementations

impl TopicBasedRemoteLogMetadataManager

pub async fn start( log: Arc<dyn MetadataEventLog>, runtime: Handle, snapshot_dir: PathBuf, snapshot_interval: Duration, ) -> Result<Arc<Self>, RemoteStorageError>

Load any on-disk snapshot into the cache and spawn the consumer pump with an empty assignment. The manager consumes nothing until Self::reconcile_assignment is driven (by the broker).

runtime must be a Tokio runtime handle that lives at least as long as the returned manager. The synchronous RemoteLogMetadataManager methods bridge to this handle via block_on, so they must NOT be called from a task running on this same runtime — the broker only invokes them through spawn_blocking, which is the only supported call pattern.

Errors

Currently infallible (the consumed set starts empty), but returns a Result so the bootstrap contract stays stable if start regains a fallible step.

pub fn shutdown(&self)

Cancel the consumer pump. Read methods continue to work against whatever was applied before shutdown; mutation methods will time out / fail to make progress.

pub async fn shutdown_and_flush(&self)

Cancel the pump + snapshotter, then write a final snapshot capturing everything applied so far. Safe to call once on graceful shutdown.

pub fn committed_offset(&self, partition: i32) -> i64

Committed offset loaded from the snapshot for a single metadata partition, or -1 when the partition is out of range or had no committed event (full replay). The assignment reconciler uses this to start a dynamically-added partition at committed + 1.

pub fn assigned_metadata_partitions(&self) -> Vec<i32>

The metadata partitions this manager is currently assigned (tracked for readiness). Sorted ascending.

pub async fn reconcile_assignment(&self, desired: &[i32])

Diff desired against the current assignment and drive the AssignmentHandle: add newly-needed partitions (seeded from the snapshot committed offset + 1, falling back to 0 when there is no committed event) and remove ones no longer needed. Records each added partition’s assignment-time HWM so reads gate on NotReady until the pump catches up.

HWM-fetch failure fails CLOSED: a partition whose real high-water mark could not be obtained is recorded with the HWM_UNKNOWN sentinel target so the gate returns NotReady (retryable), never a false Ok(None). Such partitions are re-attempted on every subsequent reconcile (which the broker drives on each image change / reconciler tick), so a transient high_water_marks failure self-heals: the sentinel is replaced with the real target as soon as the fetch succeeds.

MUST be driven by a SINGLE task. This method is not internally serialized — it interleaves .await points with reads/writes of the ready_targets map under short, non-overlapping locks — so two concurrent callers could race the add/remove/refresh logic. Correctness relies on the broker invoking it from exactly one reconciler task.

Async because it reads the log’s high-water marks; the broker calls it from its reconciler task (on the runtime), never from a spawn_blocking thread.

Trait Implementations

impl Drop for TopicBasedRemoteLogMetadataManager

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl RemoteLogMetadataManager for TopicBasedRemoteLogMetadataManager

fn add_remote_log_segment_metadata( &self, metadata: RemoteLogSegmentMetadata, ) -> Result<(), RemoteStorageError>

Record a newly-started segment copy. The metadata’s state must be CopySegmentStarted and its id must not already be known.

fn update_remote_log_segment_metadata( &self, update: RemoteLogSegmentMetadataUpdate, ) -> Result<(), RemoteStorageError>

Advance a known segment’s lifecycle state.

fn remote_log_segment_metadata( &self, topic_id_partition: &TopicIdPartition, leader_epoch: i32, offset: i64, ) -> Result<Option<RemoteLogSegmentMetadata>, RemoteStorageError>

Find the finished segment that contains offset for the given leader_epoch, if any.

fn highest_offset_for_epoch( &self, topic_id_partition: &TopicIdPartition, leader_epoch: i32, ) -> Result<Option<i64>, RemoteStorageError>

The highest offset that the remote tier holds for leader_epoch (the max end_offset across finished segments carrying that epoch).

fn list_remote_log_segments( &self, topic_id_partition: &TopicIdPartition, ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError>

All segments known for a partition, ordered by start_offset (regardless of state).

fn list_remote_log_segments_by_epoch( &self, topic_id_partition: &TopicIdPartition, leader_epoch: i32, ) -> Result<Vec<RemoteLogSegmentMetadata>, RemoteStorageError>

All segments carrying leader_epoch, ordered by start_offset (regardless of state).

fn put_remote_partition_delete_metadata( &self, metadata: RemotePartitionDeleteMetadata, ) -> Result<(), RemoteStorageError>

Record a partition-delete lifecycle event.