Struct TopicBasedRemoteLogMetadataManager 

pub struct TopicBasedRemoteLogMetadataManager { /* private fields */ }

Production RemoteLogMetadataManager backed by the __remote_log_metadata topic through a MetadataEventLog adapter.

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

Implementations

impl TopicBasedRemoteLogMetadataManager

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.

This method adds newly-needed partitions, seeded from the snapshot committed offset + 1, or from 0 when there is no committed event. It removes partitions that are no longer needed. It records each added partition’s assignment-time HWM, so reads gate on NotReady until the pump catches up.

An HWM-fetch failure fails CLOSED. This method records a partition whose real high-water mark it could not get with the HWM_UNKNOWN sentinel target, so the gate returns the retryable NotReady and never a false Ok(None). Every subsequent reconcile retries such partitions, and the broker drives a reconcile on each image change and each reconciler tick. A transient high_water_marks failure therefore self-heals: the real target replaces the sentinel as soon as the fetch succeeds.

A SINGLE task MUST drive this method. It is not internally serialized. It interleaves .await points with reads and writes of the ready_targets map under short, non-overlapping locks, so two concurrent callers could race the add, remove, and refresh logic. Correctness depends on the broker invoking it from exactly one reconciler task.

It is 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.

impl TopicBasedRemoteLogMetadataManager

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 and so needs a full replay. The assignment reconciler uses this to start a dynamically-added partition at committed + 1.

impl TopicBasedRemoteLogMetadataManager

pub fn metadata_partition_readiness(&self) -> Vec<MetadataPartitionReadiness>

One MetadataPartitionReadiness for each metadata partition this manager is assigned, sorted by partition. An unassigned partition has no entry.

ready comes from the same rule that every gated read applies, over the target and applied offset the entry reports.

impl TopicBasedRemoteLogMetadataManager

pub fn metadata_partition_count(&self) -> i32

Actual partition count of the opened metadata topic. An existing topic’s count takes precedence over its provisioning configuration.

pub 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 the broker drives Self::reconcile_assignment.

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 with block_on, so a caller must NOT call them from a task that runs on this same runtime. The broker invokes them only through spawn_blocking, which is the only supported call pattern.

Errors

Returns an error when the event log reports a negative partition count. Nothing else fails, because the consumed set starts empty.

pub fn shutdown(&self)

Cancel the consumer pump. Read methods continue to work against whatever the pump applied before shutdown. Pending and subsequent mutation methods return a closed-log error.

pub async fn shutdown_and_flush(&self)

Cancel the pump and the snapshotter, then write a final snapshot that captures everything applied so far. It is safe to call this once on a graceful shutdown.

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>

Records 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>

Advances a known segment’s lifecycle state.

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

Finds 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: LeaderEpoch, ) -> Result<Option<i64>, RemoteStorageError>

The highest offset that the remote tier holds for leader_epoch.

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

All segments known for a partition, in start_offset order. The state does not matter.

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

All segments that carry leader_epoch, in start_offset order. The state does not matter.

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

Records a partition-delete lifecycle event.

fn caught_up( &self, topic_id_partition: &TopicIdPartition, ) -> Result<bool, RemoteStorageError>

Whether this manager has applied every metadata event for topic_id_partition that the metadata store had committed when the call began.