Trait SourceConnector 

pub trait SourceConnector: Send {
    // 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;
    fn contract(
        &self,
        config: &ConnectorConfig,
    ) -> Result<SourceContract, ConnectorError>;

    // Provided methods
    fn cancellation_policy(&self) -> ConnectorCancellationPolicy { ... }
    fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker> { ... }
    fn supports_initialized_start(&self) -> bool { ... }
    fn resolve_schema<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        config: &'life1 ConnectorConfig,
        explicit: Option<SchemaRef>,
    ) -> Pin<Box<dyn Future<Output = Result<SchemaBinding, ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    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 resolve_initial_position<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        _config: &'life1 ConnectorConfig,
    ) -> Pin<Box<dyn Future<Output = Result<SourceCheckpoint, ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait { ... }
    fn validate_initial_position<'life0, 'life1, 'life2, 'async_trait>(
        &'life0 mut self,
        _config: &'life1 ConnectorConfig,
        _checkpoint: &'life2 SourceCheckpoint,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait,
             'life2: 'async_trait { ... }
    fn try_checkpoint(&self) -> Result<Option<SourceCheckpoint>, ConnectorError> { ... }
    fn checkpoint_unavailable_policy(&self) -> SourceCheckpointUnavailablePolicy { ... }
    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 recovery_identity_options(
        &self,
        _config: &ConnectorConfig,
    ) -> Result<Option<BTreeMap<String, String>>, ConnectorError> { ... }
    fn recency(
        &self,
        _config: &ConnectorConfig,
    ) -> Result<Option<Arc<dyn SourceRecency>>, 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 { ... }
}

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

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.

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.

fn schema(&self) -> SchemaRef

Arrow schema of records this source produces.

fn checkpoint(&self) -> SourceCheckpoint

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

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.

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

Declare recovery and placement semantics for this exact configuration. This hook must be deterministic, configuration-only, and free of external I/O; admission and checkpoint identity inspect it before start().

Errors

Returns an error when the concrete configuration cannot provide a valid recovery, placement, or input contract.

Provided Methods

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.

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 crate::connector::ConnectorTaskOwner and move a guard into every task. The runtime can then wait for true terminal completion after dropping the connector.

fn supports_initialized_start(&self) -> bool

Whether atomic start validates and installs SourcePosition::Initialized without resolving its boundary again or acknowledging pre-boundary input. Discovery/validation hooks alone do not certify startup. The runtime rejects unrecognized implementations before I/O; this capability is neither topology authority nor actor readiness.

fn resolve_schema<'life0, 'life1, 'async_trait>( &'life0 mut self, config: &'life1 ConnectorConfig, explicit: Option<SchemaRef>, ) -> Pin<Box<dyn Future<Output = Result<SchemaBinding, ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Resolve an immutable reader contract before activation. Metadata adapters must override this hook to retain native identity/definitions and validate projections. The default preserves explicit columns and reuses pre-open discovery for fixed schemas.

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 connector and format 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.

fn resolve_initial_position<'life0, 'life1, 'async_trait>( &'life0 mut self, _config: &'life1 ConnectorConfig, ) -> Pin<Box<dyn Future<Output = Result<SourceCheckpoint, ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Resolve a complete, unowned initial cursor without starting or consuming the source. The topology control path seals the first successful vector and reuses it on every retry. Implementations must bound metadata I/O and task ownership, avoid group joins, external acknowledgements and resource creation, and return the existing connector cursor encoding. The cursor is not a checkpoint of processed data or an intake permit. Ordinary startup and recovery keep their current policy until target installation explicitly consumes this cut.

Errors

Rejects configurations without a certified initialization contract. The default fails closed; adding a connector requires an implementation and recovery/activation evidence.

fn validate_initial_position<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, _config: &'life1 ConnectorConfig, _checkpoint: &'life2 SourceCheckpoint, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Read-only validation of a previously sealed global new-source cursor. Must preserve its numeric boundary and create no active reader, acknowledgement or sink effect. Connectors must reject a changed inventory or expired position rather than silently reset. This is preparation only; startup must validate again under final installation authority.

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.

fn checkpoint_unavailable_policy(&self) -> SourceCheckpointUnavailablePolicy

Choose how the runtime progresses after Self::try_checkpoint returns Ok(None).

The default holds intake because polling across an unreconciled control-plane cursor could move data past the retained barrier. The result must remain stable after the connector starts.

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.

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.

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.

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.

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.

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 crate::connector::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.

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).

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.

fn recency( &self, _config: &ConnectorConfig, ) -> Result<Option<Arc<dyn SourceRecency>>, ConnectorError>

How far the upstream of a source configured by config has been written, for reads that must include it. Configuration only, no external I/O. The default, None, means the connector cannot tell, and such reads are refused.

Errors

Returns a configuration error when config is invalid for this connector.

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

impl SourceConnector for DeltaSource

impl SourceConnector for FileSource

impl SourceConnector for FixSource

impl SourceConnector for GeneratorSource

impl SourceConnector for IbmMqSource

impl SourceConnector for IcebergSource

impl SourceConnector for KafkaSource

impl SourceConnector for MockSourceConnector

impl SourceConnector for MongoDbCdcSource

impl SourceConnector for NatsSource

impl SourceConnector for OtelSource

impl SourceConnector for PostgresCdcSource

impl SourceConnector for RabbitMqSource