Struct GlobalBarrierWorkerContextImpl

pub(super) struct GlobalBarrierWorkerContextImpl {
    scheduled_barriers: ScheduledBarriers,
    status: Arc<ArcSwap<BarrierManagerStatus>>,
    pub(super) metadata_manager: MetadataManager,
    hummock_manager: HummockManagerRef,
    source_manager: SourceManagerRef,
    scale_controller: ScaleControllerRef,
    pub(super) env: MetaSrvEnv,
}

Fields

scheduled_barriers: ScheduledBarriers

status: Arc<ArcSwap<BarrierManagerStatus>>

metadata_manager: MetadataManager

hummock_manager: HummockManagerRef

source_manager: SourceManagerRef

scale_controller: ScaleControllerRef

env: MetaSrvEnv

Implementations

impl GlobalBarrierWorkerContextImpl

fn set_status(&self, new_status: BarrierManagerStatus)

impl GlobalBarrierWorkerContextImpl

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

Clean catalogs for creating streaming jobs that are in foreground mode or table fragments not persisted.

async fn purge_state_table_from_hummock( &self, all_state_table_ids: &HashSet<TableId>, ) -> MetaResult<()>

async fn recover_background_mv_progress( &self, ) -> MetaResult<HashMap<TableId, (String, TableFragments)>>

async fn resolve_graph_info( &self, ) -> MetaResult<HashMap<DatabaseId, InflightDatabaseInfo>>

Resolve actor information from cluster, fragment manager and ChangedTableId. We use changed_table_id to modify the actors to be sent or collected. Because these actor will create or drop before this barrier flow through them.

pub(super) async fn reload_runtime_info_impl( &self, ) -> MetaResult<BarrierWorkerRuntimeInfoSnapshot>

impl GlobalBarrierWorkerContextImpl

const RECOVERY_FORCE_MIGRATION_TIMEOUT: Duration = _

async fn migrate_actors( &self, active_nodes: &mut ActiveStreamingWorkerNodes, ) -> MetaResult<HashMap<DatabaseId, InflightDatabaseInfo>>

Migrate actors in expired CNs to newly joined ones, return true if any actor is migrated.

async fn scale_actors( &self, active_nodes: &ActiveStreamingWorkerNodes, ) -> MetaResult<()>

fn derive_target_parallelism( available_parallelism: usize, assigned_parallelism: TableParallelism, actual_fragment_parallelism: Option<usize>, default_parallelism: DefaultParallelism, ) -> TableParallelism

async fn load_all_actors(&self) -> MetaResult<HashMap<ActorId, StreamActor>>

Update all actors in compute nodes.

impl GlobalBarrierWorkerContextImpl

pub(super) fn new( scheduled_barriers: ScheduledBarriers, status: Arc<ArcSwap<BarrierManagerStatus>>, metadata_manager: MetadataManager, hummock_manager: HummockManagerRef, source_manager: SourceManagerRef, scale_controller: ScaleControllerRef, env: MetaSrvEnv, ) -> Self

pub(super) fn status(&self) -> Arc<ArcSwap<BarrierManagerStatus>>

impl GlobalBarrierWorkerContextImpl

pub(super) async fn new_control_stream_impl( &self, node: &WorkerNode, init_request: &PbInitRequest, ) -> MetaResult<StreamingControlHandle>

Trait Implementations

impl GlobalBarrierWorkerContext for GlobalBarrierWorkerContextImpl

async fn commit_epoch( &self, commit_info: CommitEpochInfo, ) -> MetaResult<HummockVersionStats>

async fn next_scheduled(&self) -> Scheduled

fn abort_and_mark_blocked(&self, recovery_reason: RecoveryReason)

fn mark_ready(&self)

async fn post_collect_command<'a>( &'a self, command: &'a CommandContext, ) -> MetaResult<()>

async fn notify_creating_job_failed(&self, err: &MetaError)

async fn finish_creating_job(&self, job: TrackingJob) -> MetaResult<()>

async fn new_control_stream( &self, node: &WorkerNode, init_request: &PbInitRequest, ) -> MetaResult<StreamingControlHandle>

async fn reload_runtime_info( &self, ) -> MetaResult<BarrierWorkerRuntimeInfoSnapshot>