Struct GlobalBarrierWorker

pub(super) struct GlobalBarrierWorker<C> {
    enable_recovery: bool,
    periodic_barriers: PeriodicBarriers,
    in_flight_barrier_nums: usize,
    pub(super) context: Arc<C>,
    env: MetaSrvEnv,
    checkpoint_control: CheckpointControl,
    completing_task: CompletingTask,
    request_rx: UnboundedReceiver<BarrierManagerRequest>,
    active_streaming_nodes: ActiveStreamingWorkerNodes,
    sink_manager: SinkCoordinatorManager,
    control_stream_manager: ControlStreamManager,
}

crate::barrier::worker::GlobalBarrierWorker sends barriers to all registered compute nodes and collect them, with monotonic increasing epoch numbers. On compute nodes, LocalBarrierManager in risingwave_stream crate will serve these requests and dispatch them to source actors.

Configuration change in our system is achieved by the mutation in the barrier. Thus, crate::barrier::worker::GlobalBarrierWorker provides a set of interfaces like a state machine, accepting crate::barrier::command::Command that carries info to build Mutation. To keep the consistency between barrier manager and meta store, some actions like “drop materialized view” or “create mv on mv” must be done in barrier manager transactional using crate::barrier::command::Command.

Fields

enable_recovery: bool

Enable recovery or not when failover.

periodic_barriers: PeriodicBarriers

The queue of scheduled barriers.

in_flight_barrier_nums: usize

The max barrier nums in flight

context: Arc<C>

env: MetaSrvEnv

checkpoint_control: CheckpointControl

completing_task: CompletingTask

Command that has been collected but is still completing. The join handle of the completing future is stored.

request_rx: UnboundedReceiver<BarrierManagerRequest>

active_streaming_nodes: ActiveStreamingWorkerNodes

sink_manager: SinkCoordinatorManager

control_stream_manager: ControlStreamManager

Implementations

impl GlobalBarrierWorker<GlobalBarrierWorkerContextImpl>

pub async fn new( scheduled_barriers: ScheduledBarriers, env: MetaSrvEnv, metadata_manager: MetadataManager, hummock_manager: HummockManagerRef, source_manager: SourceManagerRef, sink_manager: SinkCoordinatorManager, scale_controller: ScaleControllerRef, request_rx: UnboundedReceiver<BarrierManagerRequest>, ) -> Self

Create a new crate::barrier::worker::GlobalBarrierWorker.

pub fn start(self) -> (JoinHandle<()>, Sender<()>)

async fn take_pause_on_bootstrap(&mut self) -> MetaResult<bool>

Check whether we should pause on bootstrap from the system parameter and reset it.

async fn run(self, shutdown_rx: Receiver<()>)

Start an infinite loop to take scheduled barriers and send them.

async fn run_inner(self, shutdown_rx: Receiver<()>)

impl<C: GlobalBarrierWorkerContext> GlobalBarrierWorker<C>

pub async fn clear_on_err(&mut self, err: &MetaError)

We need to make sure there are no changes when doing recovery

impl<C: GlobalBarrierWorkerContext> GlobalBarrierWorker<C>

async fn failure_recovery(&mut self, err: MetaError)

Set barrier manager status.

async fn adhoc_recovery(&mut self)

impl<C> GlobalBarrierWorker<C>

pub(super) fn report_collect_failure(&self, error: &MetaError)

Send barrier-complete-rpc and wait for responses from all CNs

impl<C> GlobalBarrierWorker<C>

const RECOVERY_RETRY_BASE_INTERVAL: u64 = 20u64

const RECOVERY_RETRY_MAX_INTERVAL: Duration = _

fn get_retry_strategy() -> impl Iterator<Item = Duration>

Initialize a retry strategy for operation in recovery.

impl<C: GlobalBarrierWorkerContext> GlobalBarrierWorker<C>

pub async fn recovery( &mut self, paused_reason: Option<PausedReason>, err: Option<MetaError>, recovery_reason: RecoveryReason, )

Recovery the whole cluster from the latest epoch.

If paused_reason is Some, all data sources (including connectors and DMLs) will be immediately paused after recovery, until the user manually resume them either by restarting the cluster or risectl command. Used for debugging purpose.

Returns the new state of the barrier manager after recovery.

async fn recovery_inner( &mut self, paused_reason: Option<PausedReason>, err: Option<MetaError>, )