Skip to main content

SourceConnector

laminar_connectors::connector

Trait SourceConnector 

Source 
pub trait SourceConnector: Send {

Show 19 methods    // Required methods
    fn start<'life0, 'async_trait>(
        &'life0 mut self,
        request: SourceStart,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn poll_batch<'life0, 'async_trait>(
        &'life0 mut self,
        max_records: usize,
    ) -> Pin<Box<dyn Future<Output = Result<Option<SourceBatch>, ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
    fn schema(&self) -> SchemaRef;
    fn checkpoint(&self) -> SourceCheckpoint;
    fn close<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;

    // Provided methods
    fn cancellation_policy(&self) -> ConnectorCancellationPolicy { ... }
    fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker> { ... }
    fn discover_schema<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        _properties: &'life1 HashMap<String, String>,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn try_checkpoint(&self) -> Result<Option<SourceCheckpoint>, ConnectorError> { ... }
    fn checkpoint_ready(&self) -> Result<bool, ConnectorError> { ... }
    fn drive_control_plane(&mut self) { ... }
    fn begin_drain(
        &mut self,
        _request: &SourceDrainRequest,
        _deadline: Instant,
    ) -> Result<(), ConnectorError> { ... }
    fn poll_drain_ready(
        &mut self,
        _round: AssignmentDrainId,
    ) -> Result<bool, ConnectorError> { ... }
    fn finish_drain<'life0, 'async_trait>(
        &'life0 mut self,
        _resolution: SourceDrainResolution,
        _deadline: Instant,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn set_vnode_assignment(
        &mut self,
        _source_identity: &str,
        _registry: Arc<VnodeRegistry>,
        _self_id: NodeId,
    ) -> Result<(), ConnectorError> { ... }
    fn data_ready_notify(&self) -> Option<Arc<Notify>> { ... }
    fn contract(
        &self,
        _config: &ConnectorConfig,
    ) -> Result<SourceContract, ConnectorError> { ... }
    fn recovery_identity_options(
        &self,
        _config: &ConnectorConfig,
    ) -> Result<Option<BTreeMap<String, String>>, ConnectorError> { ... }
    fn notify_epoch_committed<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        _epoch: u64,
        _checkpoint: &'life1 SourceCheckpoint,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }

}
Expand description

Trait for source connectors that read data from external systems.

Source connectors operate in Ring 1 and push data into Ring 0 via the streaming Source<ArrowRecord>::push_arrow() API.

§Lifecycle

  1. start() — atomically install the configured or recovered cursor and initialize the reader
  2. poll_batch() — read batches in a loop
  3. checkpoint() — capture the current connector cursor
  4. close() — clean shutdown

Required Methods§

Source

fn start<'life0, 'async_trait>( &'life0 mut self, request: SourceStart, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Opens the source and establishes its initial or resumed position as one indivisible lifecycle transition.

Implementations must not emit records or expose an externally active consumer before the requested position has been applied successfully.

Source

fn poll_batch<'life0, 'async_trait>( &'life0 mut self, max_records: usize, ) -> Pin<Box<dyn Future<Output = Result<Option<SourceBatch>, ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Ok(None) = no data currently available; runtime retries after a delay. max_records is the normal batching target. A source may exceed it only when one upstream atomic replay unit cannot be split without making its checkpoint cursor invalid. Such sources must enforce independent hard record and byte limits and fail before retained data can grow unbounded.

The runtime may cancel this future at a shutdown or authority deadline. ConnectorCancellationPolicy::CancelSafe implementations must not advance external or checkpoint-visible position across an .await unless dropping the future there preserves replay of every not-yet-returned record. Stage work privately, then advance the cursor and return without another cancellation point. The conservative ConnectorCancellationPolicy::RetireConnector policy instead makes the complete connector generation terminal after cancellation.

Source

fn schema(&self) -> SchemaRef

Arrow schema of records this source produces.

Source

fn checkpoint(&self) -> SourceCheckpoint

Returned checkpoint must contain enough info to resume from the current position after a restart.

Source

fn close<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Close the connection and release resources.

Provided Methods§

Source

fn cancellation_policy(&self) -> ConnectorCancellationPolicy

Deadline behavior required by the underlying client implementation.

Retirement is the conservative default: a new connector must not be reused after cancellation until every lifecycle future has been audited.

Source

fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker>

Observe detached tasks whose lifetime may outlast this connector value.

A connector that spawns detached work must retain the matching ConnectorTaskOwner and move a guard into every task. The runtime can then wait for true terminal completion after dropping the connector.

Source

fn discover_schema<'life0, 'life1, 'async_trait>( &'life0 mut self, _properties: &'life1 HashMap<String, String>, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Resolve the source schema from the WITH (...) properties before DDL reaches the planner. Implementations that perform network I/O must bound it with a timeout. Return Err(ConnectorError::…) on failure so the runtime can surface the cause to DDL — do not log and swallow.

Source

fn try_checkpoint(&self) -> Result<Option<SourceCheckpoint>, ConnectorError>

Atomically attempt to capture a source cursor. Ok(None) means a transient control-plane publication is not reconciled yet; the runtime retains the barrier and retries without advancing it.

§Errors

Returns a connector error when the source cannot produce a valid cursor for the current ownership/configuration state.

Source

fn checkpoint_ready(&self) -> Result<bool, ConnectorError>

Whether this source’s data-plane cursor is reconciled with the current control-plane ownership publication. The runtime does not poll records or admit checkpoint barriers while this is false.

Non-partitioned and local sources are always ready. Cluster-aware sources override this with a lock-free version fence.

§Errors

Returns a connector error when control-plane reconciliation detects invalid or lost ownership state that cannot be retried safely.

Source

fn drive_control_plane(&mut self)

Start or advance connector-side control-plane work needed to become checkpoint_ready. Called even while data polling and barriers are fenced.

Source

fn begin_drain( &mut self, _request: &SourceDrainRequest, _deadline: Instant, ) -> Result<(), ConnectorError>

Stop advancing external input for an exact cluster transition.

Implementations start the provider operation without blocking and later expose readiness through Self::poll_drain_ready. deadline is the engine-owned absolute deadline for this drain attempt; provider retries must not continue beyond it. Actor-only sources are fenced by the engine and never call this hook; assignment-scoped sources must implement it explicitly.

§Errors

Returns an error when the provider cannot start this exact drain round.

Source

fn poll_drain_ready( &mut self, _round: AssignmentDrainId, ) -> Result<bool, ConnectorError>

Whether the exact provider FIFO boundary has been consumed.

Returns false while the reader is still pausing or while pre-boundary payloads remain.

§Errors

Returns an error when drain progress cannot be observed safely.

Source

fn finish_drain<'life0, 'async_trait>( &'life0 mut self, _resolution: SourceDrainResolution, _deadline: Instant, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Resolve an exact drain after target commit or abort.

An abort must rewind any client-delivered but engine-unaccepted records before resuming. A commit must reconcile target ownership before clearing its post-cut filter. deadline is the engine-owned absolute deadline for the complete resolution; provider retries and blocking client calls must be bounded by its remaining time.

Source

fn set_vnode_assignment( &mut self, _source_identity: &str, _registry: Arc<VnodeRegistry>, _self_id: NodeId, ) -> Result<(), ConnectorError>

Install the cluster vnode assignment for a source that advertises SourceTopology::Splittable. The source identity is the stable, canonical catalog object name and must be part of any external split mapping ABI.

Embedded, single-node, singleton, and node-local sources are not sent this hook. The default fails closed so an extension cannot advertise splittable placement while every cluster node reads the full input.

§Errors

Returns a configuration error unless the connector implements exact vnode-scoped input ownership.

Source

fn data_ready_notify(&self) -> Option<Arc<Notify>>

Returns a [Notify] handle that is signalled when new data is available.

When Some, the pipeline coordinator awaits the notification instead of polling on a timer, eliminating idle CPU usage. Push-driven sources should return Some and call notify.notify_one() when data arrives.

The default implementation returns None, which causes the pipeline to fall back to timer-based polling (suitable for batch/file sources).

Source

fn contract( &self, _config: &ConnectorConfig, ) -> Result<SourceContract, ConnectorError>

Declare recovery and placement semantics for this exact configuration.

The fail-closed default is an ephemeral singleton. Durable or distributed semantics must be opted into by the connector explicitly.

§Errors

Returns an error when the concrete configuration cannot provide a valid recovery or placement contract. The default implementation never fails.

Source

fn recovery_identity_options( &self, _config: &ConnectorConfig, ) -> Result<Option<BTreeMap<String, String>>, ConnectorError>

Return connector-owned semantic options for durable recovery identity.

The hook must be deterministic, configuration-only, and free of external I/O. Some replaces the raw property map in the pipeline identity; None asks the runtime to use its conservative sanitized-property fallback. A connector may omit operational endpoints or credentials only when its checkpoint independently binds the exact external object.

§Errors

Returns a configuration error when the semantic identity cannot be derived from the supplied source configuration.

Source

fn notify_epoch_committed<'life0, 'life1, 'async_trait>( &'life0 mut self, _epoch: u64, _checkpoint: &'life1 SourceCheckpoint, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Acknowledge that epoch has been durably committed.

Called after the manifest and exact engine commit decision are durable. Coordinated external sink publication may complete asynchronously afterward. The checkpoint is the exact per-source SourceCheckpoint that was persisted into the manifest for this epoch — sources can rely on it to advance external offset state (broker group offsets, lookup-DB cursors, ack tokens) using values that match what’s durable.

May be called with an empty checkpoint for timer-driven commits where no per-source state was captured; implementations should treat that as a no-op for any externally-visible advancement.

Idempotent — a retry after cancellation is legal.

§Errors

The epoch is already durable and cannot be rolled back. During normal processing an error faults replay-capable pipelines so recovery retries the advisory upstream commit; during shutdown it is logged and the durable LaminarDB checkpoint remains authoritative.

Dyn Compatibility§

This trait is dyn compatible.

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

Implementors§