Trait PipelineCallback 

pub trait PipelineCallback:
    Send
    + 'static
    + Send {
    // Required methods
    fn execute_cycle(
        &mut self,
        source_batches: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
        watermark: i64,
    ) -> impl Future<Output = Result<CycleOutcome, CycleError>> + Send;
    fn drain_checkpoint_edges_until(
        &mut self,
        deadline: Instant,
    ) -> impl Future<Output = Result<(), CycleError>> + Send;
    fn write_to_sinks(
        &mut self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
        deadline: Option<Instant>,
    ) -> impl Future<Output = Result<(), CycleError>> + Send;
    fn extract_watermark(&mut self, source_name: &str, batch: &RecordBatch);
    fn filter_late_rows(
        &self,
        source_name: &str,
        batch: &RecordBatch,
    ) -> Option<RecordBatch>;
    fn current_watermark(&self) -> i64;
    fn abandon_checkpoint_attempt(
        &mut self,
        _attempt: CheckpointAttempt,
        _reason: &str,
        _assignment_fence: Option<CheckpointAssignmentFence>,
    ) -> impl Future<Output = Result<(), String>> + Send;
    fn cancel_source_barrier_attempt(
        &mut self,
        _attempt: CheckpointAttempt,
        _reason: &str,
    ) -> impl Future<Output = Result<(), String>> + Send;
    fn service_checkpoint_control(
        &mut self,
        source_offsets: FxHashMap<String, SourceCheckpoint>,
    ) -> impl Future<Output = CheckpointControlOutcome> + Send;
    fn checkpoint_with_barrier(
        &mut self,
        source_checkpoints: FxHashMap<String, SourceCheckpoint>,
        attempt: CheckpointAttempt,
        attempt_started: Instant,
        assignment_fence: Option<CheckpointAssignmentFence>,
    ) -> impl Future<Output = BarrierOutcome> + Send;
    fn record_cycle(&self, events_ingested: u64, batches: u64, elapsed_ns: u64);
    fn apply_control(&mut self, msg: ControlMsg);

    // Provided methods
    fn prepare_source_intake(&mut self) -> Result<(), String> { ... }
    fn push_to_streams(
        &self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
    ) -> Result<(), CycleError> { ... }
    fn update_mv_stores(
        &self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
    ) -> Result<(), CycleError> { ... }
    fn is_leader(&self) -> bool { ... }
    fn is_recovering(&mut self) -> bool { ... }
    fn fault_on_cycle_error(&self) -> bool { ... }
    fn take_pipeline_fault(&mut self) -> Option<String> { ... }
    fn record_checkpoint_failure(&mut self, _checkpoint_id: u64, _reason: &str) { ... }
    fn record_checkpoint_continuation_fault(
        &mut self,
        _attempt: CheckpointAttempt,
        _reason: &str,
    ) { ... }
    fn record_checkpoint_admission_failure(&mut self, _reason: &str) { ... }
    fn settle_checkpoint_tail_tasks(
        &mut self,
        _abort: bool,
    ) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn reserve_checkpoint_attempt(
        &mut self,
        _attempt_started: Instant,
    ) -> impl Future<Output = Result<CheckpointAttempt, String>> + Send { ... }
    fn publish_checkpoint_prepare(
        &mut self,
        _attempt: CheckpointAttempt,
        _attempt_started: Instant,
        _assignment_fence: Option<CheckpointAssignmentFence>,
    ) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn resolve_authoritative_follower_abort(
        &mut self,
        _attempt: CheckpointAttempt,
    ) -> Result<(), String> { ... }
    fn checkpoint_assignment_for_admission(
        &mut self,
    ) -> impl Future<Output = CheckpointAssignmentAdmission> + Send { ... }
    fn checkpoint_control_wake(&self) -> Option<CheckpointControlWake> { ... }
    fn tick_idle_watermark(&mut self) { ... }
    fn note_cycle_error(&self) { ... }
    fn is_backpressured(&self) -> bool { ... }
    fn intake_paused(&self) -> bool { ... }
    fn has_deferred_input(&self) -> bool { ... }
    fn reserve_subscription_cut(
        &self,
        _attempt: CheckpointAttempt,
    ) -> Result<(), String> { ... }
    fn abort_subscription_cut(&self, _attempt: CheckpointAttempt) { ... }
    fn publish_barrier(&self, _attempt: CheckpointAttempt) -> Result<(), String> { ... }
    fn invalidate_subscriptions(&self, _reason: &str) { ... }
    fn settle_sink_epoch_for_shutdown(
        &mut self,
    ) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn close_sinks(&mut self) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn set_barrier_injectors(&mut self, injectors: Vec<SourceBarrierControl>) { ... }
}

Callback trait for the coordinator to interact with the rest of the DB. Trait exists for test seam; production impl is ConnectorPipelineCallback.

Required Methods

fn execute_cycle( &mut self, source_batches: &FxHashMap<Arc<str>, Vec<RecordBatch>>, watermark: i64, ) -> impl Future<Output = Result<CycleOutcome, CycleError>> + Send

Execute a SQL cycle over the accumulated source batches. Err is a whole-cycle failure (all domains, or a backpressure halt); per-domain faults surface in CycleOutcome so healthy domains still commit.

fn drain_checkpoint_edges_until( &mut self, deadline: Instant, ) -> impl Future<Output = Result<(), CycleError>> + Send

Drain every graph input that belongs to the frozen checkpoint cut.

Implementations must deliver each drain pass’s outputs before returning and must not cancel an in-progress graph pass: operators may temporarily own their input buffers across an await. The absolute deadline is checked between complete passes.

fn write_to_sinks( &mut self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, deadline: Option<Instant>, ) -> impl Future<Output = Result<(), CycleError>> + Send

Write cycle results to sinks, bounded by deadline when this is a checkpoint drain.

fn extract_watermark(&mut self, source_name: &str, batch: &RecordBatch)

Extract watermark from a batch for a given source.

fn filter_late_rows( &self, source_name: &str, batch: &RecordBatch, ) -> Option<RecordBatch>

Filter late rows from a batch.

fn current_watermark(&self) -> i64

Current pipeline watermark.

fn abandon_checkpoint_attempt( &mut self, _attempt: CheckpointAttempt, _reason: &str, _assignment_fence: Option<CheckpointAssignmentFence>, ) -> impl Future<Output = Result<(), String>> + Send

Abandon a reserved attempt that cannot reach the durable checkpoint tail. Transactional sinks must roll back that epoch before the next attempt begins.

fn cancel_source_barrier_attempt( &mut self, _attempt: CheckpointAttempt, _reason: &str, ) -> impl Future<Output = Result<(), String>> + Send

Cancel an exact follower source-barrier attempt before capture.

An attempt admitted with follower ownership releases its exact local reservation and publishes a negative barrier acknowledgement, even if this process changes role while the attempt is active. Originator-owned attempts use abandon_checkpoint_attempt.

fn service_checkpoint_control( &mut self, source_offsets: FxHashMap<String, SourceCheckpoint>, ) -> impl Future<Output = CheckpointControlOutcome> + Send

Service a cluster follower announcement observed from the leader.

Periodic, manual, and shutdown admission belongs exclusively to the streaming coordinator. This control seam must never originate a local checkpoint.

fn checkpoint_with_barrier( &mut self, source_checkpoints: FxHashMap<String, SourceCheckpoint>, attempt: CheckpointAttempt, attempt_started: Instant, assignment_fence: Option<CheckpointAssignmentFence>, ) -> impl Future<Output = BarrierOutcome> + Send

Called when all sources have aligned on a barrier.

fn record_cycle(&self, events_ingested: u64, batches: u64, elapsed_ns: u64)

Record cycle metrics.

fn apply_control(&mut self, msg: ControlMsg)

Apply a DDL control message (add/drop stream) to the running pipeline.

Provided Methods

fn prepare_source_intake(&mut self) -> Result<(), String>

Install any newly published recovery cut before the coordinator removes another source message from its FIFO. The default is a no-op outside a clustered source-handoff runtime.

Errors

Returns an error if a pending recovery cut cannot be installed.

fn push_to_streams( &self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, ) -> Result<(), CycleError>

Push cycle results to stream subscriptions.

Errors

Returns an error if subscription delivery rejects the cycle output.

fn update_mv_stores( &self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, ) -> Result<(), CycleError>

Update materialized view stores with cycle results.

Errors

Returns an error if materialized state cannot apply the cycle output.

fn is_leader(&self) -> bool

true if this node is the cluster leader, or in single-node mode.

fn is_recovering(&mut self) -> bool

true while a coordinated restart is in flight; the checkpoint admission gate holds. Default false.

fn fault_on_cycle_error(&self) -> bool

true if a fatal cycle error should fault for recovery rather than drop-and-continue (exactly-once, or coordinated recovery). Default false (at-least-once drops).

fn take_pipeline_fault(&mut self) -> Option<String>

Take a pending consistency fault from checkpointing or a poisoned sink epoch. The coordinator stops intake so recovery can replay from the last committed cut.

fn record_checkpoint_failure(&mut self, _checkpoint_id: u64, _reason: &str)

Record a checkpoint failure observed by the coordinator. Exactly-once implementations fault for recovery; weaker guarantees may retain retry-on-next-interval behaviour.

fn record_checkpoint_continuation_fault( &mut self, _attempt: CheckpointAttempt, _reason: &str, )

Record an invariant failure after the checkpoint itself became durable.

fn record_checkpoint_admission_failure(&mut self, _reason: &str)

Record a failure before an exact checkpoint attempt could be reserved.

fn settle_checkpoint_tail_tasks( &mut self, _abort: bool, ) -> impl Future<Output = Result<(), String>> + Send

Join tracked asynchronous checkpoint tails before connector teardown. When abort is true, request cancellation and detach them because the bounded graceful-drain budget has expired and cancellation may be cooperative.

fn reserve_checkpoint_attempt( &mut self, _attempt_started: Instant, ) -> impl Future<Output = Result<CheckpointAttempt, String>> + Send

Durably reserve the exact attempt before barriers are admitted to sources.

Implementations must never synthesize an in-memory checkpoint ID. A successful reservation may be abandoned, but its ID is permanently burned.

fn publish_checkpoint_prepare( &mut self, _attempt: CheckpointAttempt, _attempt_started: Instant, _assignment_fence: Option<CheckpointAssignmentFence>, ) -> impl Future<Output = Result<(), String>> + Send

Publish the certified cluster Prepare for an exact reserved attempt before any source or shuffle barrier is injected. Local runtimes have no cluster control record.

fn resolve_authoritative_follower_abort( &mut self, _attempt: CheckpointAttempt, ) -> Result<(), String>

Resolve exact local follower state after an authoritative pre-capture BarrierOutcome::Aborted. This operation must not publish control traffic or wait on the network because cluster authority has already terminated the attempt.

Errors

Returns an error if local follower state cannot be resolved exactly.

fn checkpoint_assignment_for_admission( &mut self, ) -> impl Future<Output = CheckpointAssignmentAdmission> + Send

Capture the exact assignment certificate for a new attempt.

fn checkpoint_control_wake(&self) -> Option<CheckpointControlWake>

Wake the coordinator for leader-originated checkpoint control. None keeps local runtimes free of cluster polling.

fn tick_idle_watermark(&mut self)

Demote sources idle past their timeout so a quiet input doesn’t pin the combined watermark.

fn note_cycle_error(&self)

Count a fatal cycle error that was dropped-and-continued (at-least-once only).

fn is_backpressured(&self) -> bool

true when internal buffers are near capacity.

fn intake_paused(&self) -> bool

true while the runtime must not fold source or shuffle input into operator state. Cluster startup and coordinated recovery use this stronger fence; ordinary backpressure only pauses source polling.

fn has_deferred_input(&self) -> bool

true when deferred operators have pending input to drain.

fn reserve_subscription_cut( &self, _attempt: CheckpointAttempt, ) -> Result<(), String>

Reserve each subscription log’s cursor at the aligned checkpoint cut.

Errors

Returns an error if the subscription cut cannot be reserved atomically.

fn abort_subscription_cut(&self, _attempt: CheckpointAttempt)

Discard an unresolved subscription cut after checkpoint failure.

fn publish_barrier(&self, _attempt: CheckpointAttempt) -> Result<(), String>

Resolve the exact cut for external SUBSCRIBE consumers after durable commit.

Errors

Returns an error if the committed cut cannot be published atomically.

fn invalidate_subscriptions(&self, _reason: &str)

Terminate provisional subscription delivery before shutdown or recovery replay.

fn settle_sink_epoch_for_shutdown( &mut self, ) -> impl Future<Output = Result<(), String>> + Send

Resolve durable ownership of any open checkpoint-committable sink epoch while its actor is still live. A failure must leave the actors open so lifecycle teardown can retry.

fn close_sinks(&mut self) -> impl Future<Output = Result<(), String>> + Send

Gracefully close sinks on shutdown (abort open transactions, flush) so a restart re-initialises cleanly. Every sink must be attempted; the result aggregates failures.

fn set_barrier_injectors(&mut self, injectors: Vec<SourceBarrierControl>)

Register the local source barrier injectors.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors