Struct SourceManager

pub struct SourceManager {
    pub paused: Mutex<()>,
    barrier_scheduler: BarrierScheduler,
    core: Mutex<SourceManagerCore>,
    pub metrics: Arc<MetaMetrics>,
}

SourceManager keeps fetching the latest split metadata from the external source services (ConnectorSourceWorker::tick), and sends a split assignment command if split changes detected (Self::tick).

Fields

paused: Mutex<()>

barrier_scheduler: BarrierScheduler

core: Mutex<SourceManagerCore>

metrics: Arc<MetaMetrics>

Implementations

impl SourceManager

const DEFAULT_SOURCE_TICK_INTERVAL: Duration = _

pub async fn new( barrier_scheduler: BarrierScheduler, metadata_manager: MetadataManager, metrics: Arc<MetaMetrics>, ) -> MetaResult<Self>

pub async fn drop_source_fragments( &self, source_fragments: HashMap<SourceId, BTreeSet<FragmentId>>, removed_actors: HashSet<ActorId>, )

pub async fn drop_source_fragments_vec( &self, table_fragments: &[TableFragments], )

For dropping MV.

pub async fn apply_source_change( &self, added_source_fragments: Option<HashMap<SourceId, BTreeSet<FragmentId>>>, added_backfill_fragments: Option<HashMap<SourceId, BTreeSet<(FragmentId, FragmentId)>>>, split_assignment: Option<SplitAssignment>, dropped_actors: Option<HashSet<ActorId>>, )

Updates states after split change (post_collect barrier) or scaling (post_apply_reschedule).

pub async fn migrate_splits_for_source_actors( &self, fragment_id: FragmentId, prev_actor_ids: &[ActorId], curr_actor_ids: &[ActorId], ) -> MetaResult<HashMap<ActorId, Vec<SplitImpl>>>

Migrates splits from previous actors to the new actors for a rescheduled fragment.

Very occasionally split removal may happen during scaling, in which case we need to use the old splits for reallocation instead of the latest splits (which may be missing), so that we can resolve the split removal in the next command.

pub fn migrate_splits_for_backfill_actors( &self, fragment_id: FragmentId, upstream_fragment_ids: &Vec<FragmentId>, curr_actor_ids: &[ActorId], fragment_actor_splits: &HashMap<FragmentId, HashMap<ActorId, Vec<SplitImpl>>>, no_shuffle_upstream_actor_map: &HashMap<ActorId, HashMap<FragmentId, ActorId>>, ) -> MetaResult<HashMap<ActorId, Vec<SplitImpl>>>

Migrates splits from previous actors to the new actors for a rescheduled fragment.

pub async fn allocate_splits( &self, table_id: &TableId, ) -> MetaResult<SplitAssignment>

Allocates splits to actors for a newly created source executor.

pub async fn allocate_splits_for_backfill( &self, table_id: &TableId, dispatchers: &HashMap<ActorId, Vec<Dispatcher>>, ) -> MetaResult<SplitAssignment>

Allocates splits to actors for a newly created SourceBackfill executor.

Unlike Self::allocate_splits, which creates a new assignment, this method aligns the splits for backfill fragments with its upstream source fragment (align_backfill_splits).

pub async fn register_source(&self, source: &Source) -> MetaResult<()>

create and register connector worker for source.

pub async fn register_source_with_handle( &self, source_id: SourceId, handle: ConnectorSourceWorkerHandle, )

register connector worker for source.

pub async fn unregister_sources(&self, source_ids: Vec<SourceId>)

Unregister connector worker for source.

fn create_source_worker_async( source: Source, managed_sources: &mut HashMap<SourceId, ConnectorSourceWorkerHandle>, metrics: Arc<MetaMetrics>, ) -> MetaResult<()>

Used on startup (Self::new). Failed sources will not block meta startup.

pub async fn list_assignments(&self) -> HashMap<ActorId, Vec<SplitImpl>>

pub async fn get_running_info(&self) -> SourceManagerRunningInfo

async fn tick(&self) -> MetaResult<()>

Checks whether the external source metadata has changed, and sends a split assignment command if it has.

This is also how a newly created SourceExecutor is initialized. (force tick in Self::create_source_worker)

The command will first updates SourceExecutor’s splits, and finally calls Self::apply_source_change to update states in SourceManager.

pub async fn run(&self) -> MetaResult<()>