Skip to main content

SinkConnector

laminar_connectors::connector

Trait SinkConnector 

Source 
pub trait SinkConnector: Send {

Show 17 methods    // Required methods
    fn open<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        config: &'life1 ConnectorConfig,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn write_batch<'life0, 'life1, 'async_trait>(
        &'life0 mut self,
        batch: &'life1 RecordBatch,
    ) -> Pin<Box<dyn Future<Output = Result<WriteResult, ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn schema(&self) -> SchemaRef;
    fn suggested_write_timeout(&self) -> Duration;
    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 bind_runtime_context(
        &mut self,
        _context: SinkRuntimeContext,
    ) -> Result<(), ConnectorError> { ... }
    fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker> { ... }
    fn contract(
        &self,
        _config: &ConnectorConfig,
    ) -> Result<SinkContract, ConnectorError> { ... }
    fn begin_epoch<'life0, 'async_trait>(
        &'life0 mut self,
        _epoch: u64,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn checkpoint_artifact_intent<'life0, 'async_trait>(
        &'life0 mut self,
        _epoch: u64,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn pre_commit<'life0, 'async_trait>(
        &'life0 mut self,
        _epoch: u64,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn rollback_epoch<'life0, 'async_trait>(
        &'life0 mut self,
        _epoch: u64,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn flush_interval(&self) -> Duration { ... }
    fn flush<'life0, 'async_trait>(
        &'life0 mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait { ... }
    fn as_coordinated_committer(&self) -> Option<&dyn CoordinatedCommitter> { ... }
    fn coordinated_abort_cleaner(
        &self,
    ) -> Option<Arc<dyn CoordinatedAbortCleaner>> { ... }

}
Expand description

Trait for sink connectors that write data to external systems.

Sink connectors operate in Ring 1, receiving data from Ring 0 and writing to external systems. Implementations whose contract is crate::connector::SinkConsistency::CheckpointCommittable prepare checkpoint-owned committables with begin_epoch/pre_commit, expose a CoordinatedCommitter for the single external commit, and implement rollback_epoch; the runtime drives them via the checkpoint coordinator.

All sinks follow open() → write_batch()/flush() → close(). Checkpoint-committable sinks additionally loop over begin_epoch(), staged writes, pre_commit(), and coordinated commit (or rollback_epoch() on a proven pre-decision failure).

Required Methods§

Source

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

Open the connection and prepare to accept writes.

Source

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

Implementations using ConnectorCancellationPolicy::CancelSafe must remain valid when this future is dropped at a deadline. For ConnectorCancellationPolicy::RetireConnector, cancellation makes the complete connector instance terminal before later work can be processed.

Source

fn schema(&self) -> SchemaRef

Expected Arrow schema of input batches.

Source

fn suggested_write_timeout(&self) -> Duration

Default per-call write_batch I/O timeout. Users can override this via the sink.write.timeout.ms connector property.

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 sink 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 bind_runtime_context( &mut self, _context: SinkRuntimeContext, ) -> Result<(), ConnectorError>

Bind the checkpoint runtime identity before open.

The default is a no-op for sinks whose object naming does not depend on checkpoint identity.

§Errors

Implementations return an error when the supplied identity is invalid or cannot be applied in the connector’s current lifecycle state.

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

Source

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

Declare durability, placement, and input semantics for this exact configuration without opening files, sockets, clients, or transactions.

The fail-closed default is an ephemeral append-only singleton. Durable or distributed behaviour must be opted into explicitly.

§Errors

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

Source

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

Begin checkpoint-owned staging. Called only for an admitted checkpoint-committable contract; weaker sinks use the no-op default.

Source

fn checkpoint_artifact_intent<'life0, 'async_trait>( &'life0 mut self, _epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Return bounded recovery evidence before the runtime allows this epoch to write.

The checkpoint store persists the result before begin_epoch. It must contain no credentials and must deterministically identify only artifacts owned by this runtime context and epoch. None means the connector has no cleanup payload; it does not prove that the connector created no external artifacts.

Source

fn pre_commit<'life0, 'async_trait>( &'life0 mut self, _epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, ConnectorError>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait,

Flush + prepare, but do not finalize externally. The runtime persists the checkpoint decision before a designated committer finalizes the collected descriptors; on failure it calls rollback_epoch.

Returns an opaque commit descriptor for checkpoint-committable sinks (the committables the designated committer will aggregate), else None. Default delegates to flush() and returns None.

§Errors

Returns ConfigurationError if the sink exposes a coordinated committer yet relies on this default — it would finalize epochs with no external commit.

Source

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

Must be idempotent. The runtime calls this on every checkpoint-committable sink after proving a pre-decision failure, including sinks that never completed pre_commit.

Source

fn flush_interval(&self) -> Duration

Maximum residence time for a non-empty sink buffer before the runtime invokes flush. Checkpoint-committable sinks ignore the periodic timer and flush only through their checkpoint protocol.

Source

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

Must be internally bounded — the sink task’s periodic timer calls this on every tick. Thorough drains belong in pre_commit / coordinated commit / close, not here.

Source

fn as_coordinated_committer(&self) -> Option<&dyn CoordinatedCommitter>

Leader-side committer for a checkpoint-committable contract; None for every weaker contract.

Source

fn coordinated_abort_cleaner(&self) -> Option<Arc<dyn CoordinatedAbortCleaner>>

Detached cleanup authority retained by recovery after the participant writer closes.

None means this connector has no artifacts that require immediate cleanup after an authoritative Abort. Files intentionally delegated to a retention-safe maintenance policy do not require a cleaner here.

Dyn Compatibility§

This trait is dyn compatible.

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

Implementors§