Struct GlobalBarrierWorker 

pub(super) struct GlobalBarrierWorker<C> {
    enable_recovery: bool,
    periodic_barriers: PeriodicBarriers,
    system_enable_per_database_isolation: bool,
    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,
    term_id: String,
}

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.

system_enable_per_database_isolation: bool

Whether per database failure isolation is enabled in system parameters.

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

term_id: String

Implementations

impl<C: GlobalBarrierWorkerContext> GlobalBarrierWorker<C>

pub(super) async fn new_inner( env: MetaSrvEnv, sink_manager: SinkCoordinatorManager, request_rx: UnboundedReceiver<BarrierManagerRequest>, context: Arc<C>, ) -> Self

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>, barrier_scheduler: BarrierScheduler, ) -> 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.

impl<C: GlobalBarrierWorkerContext> GlobalBarrierWorker<C>

fn enable_per_database_isolation(&self) -> bool

pub(super) 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(env: &MetaSrvEnv, error: &MetaError)

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

impl<C: GlobalBarrierWorkerContext> GlobalBarrierWorker<C>

pub async fn recovery( &mut self, is_paused: bool, 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, is_paused: bool, recovery_reason: String)