Struct MongoDbCdcSource 

pub struct MongoDbCdcSource { /* private fields */ }

MongoDB CDC source connector.

Streams change events from a MongoDB change stream using the SourceConnector trait. Events are buffered internally and converted to Arrow RecordBatches on poll_batch.

Resume Token Handling

The connector tracks resume tokens from individual change event _id fields. Tokens are persisted via a pluggable ResumeTokenStore.

Sharded Cluster Note

On sharded clusters, mongos opens per-shard cursors and merges results transparently. Ensure max_pool_size is at least as large as the expected number of concurrent change streams to avoid connection starvation.

Implementations

impl MongoDbCdcSource

pub fn new(config: MongoDbSourceConfig, registry: Option<&Registry>) -> Self

Creates a new MongoDB CDC source with the given configuration.

pub fn from_config(config: &ConnectorConfig) -> Result<Self, ConnectorError>

Creates a new source from a generic ConnectorConfig.

Errors

Returns ConnectorError if the configuration is invalid.

pub fn with_resume_token_store(self, store: Box<dyn ResumeTokenStore>) -> Self

Sets a custom resume token store.

pub fn config(&self) -> &MongoDbSourceConfig

Returns a reference to the source configuration.

pub fn last_resume_token(&self) -> Option<&ResumeToken>

Returns the last persisted resume token.

pub fn buffered_events(&self) -> usize

Returns the number of buffered events.

pub fn is_invalidated(&self) -> bool

Returns true if the stream has been invalidated.

pub fn enqueue_event(&mut self, event: MongoDbChangeEvent)

Enqueues a change event for processing (used by tests and the background reader task).

pub fn drain_to_batch( &mut self, max_records: usize, ) -> Result<Option<SourceBatch>, ConnectorError>

Drains up to max_records events from the buffer and converts them to an Arrow RecordBatch.

Errors

Returns ConnectorError if Arrow batch construction fails.

Trait Implementations

impl SourceConnector for MongoDbCdcSource

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,

Called once before polling begins.

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 a hint — implementations may return fewer.

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 restore<'life0, 'life1, 'async_trait>( &'life0 mut self, checkpoint: &'life1 SourceCheckpoint, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

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

Called during recovery before polling resumes.

fn health_check(&self) -> HealthStatus

Defaults to Unknown; connectors should override.

fn metrics(&self) -> ConnectorMetrics

Current connector metrics snapshot.

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 data_ready_notify(&self) -> Option<Arc<Notify>>

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

fn supports_replay(&self) -> bool

Whether this source supports replay from a checkpointed position.

fn discover_schema<'life0, 'life1, 'async_trait>( &'life0 mut self, _properties: &'life1 HashMap<String, String>, ) -> Pin<Box<dyn Future<Output = ()> + 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 hit the network (e.g. Kafka fetching an Avro schema from a Schema Registry) must bound their I/O with a timeout and leave the schema empty on failure rather than hang.

fn as_schema_provider(&self) -> Option<&dyn SchemaProvider>

Returns this connector as a SchemaProvider, if supported.

fn as_schema_registry_aware(&self) -> Option<&dyn SchemaRegistryAware>

Returns this connector as a SchemaRegistryAware, if supported.

fn checkpoint_requested(&self) -> Option<Arc<AtomicBool>>

Returns a shared flag that the source sets to true when it requests an immediate checkpoint.

fn notify_epoch_committed<'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,

Acknowledge that epoch has been durably committed.