Struct KafkaSource 

pub struct KafkaSource { /* private fields */ }

Kafka source connector that consumes messages and produces Arrow batches.

Operates in Ring 1 (background) and pushes deserialized RecordBatch data to Ring 0 via the streaming Source<T> API.

Lifecycle

1. Create with KafkaSource::new or KafkaSource::with_schema_registry
2. Call start() with the initial or exact recovered position
3. Call poll_batch() in a loop to consume messages
4. Call checkpoint() for fault tolerance
5. Call close() for clean shutdown

Guaranteed delivery uses engine-owned manual assignment: the complete explicit topic inventory in embedded/single-node mode or the owned vnode subset in cluster mode. Broker consumer-group ownership can revoke a partition before an asynchronous engine checkpoint captures its final offset, so dynamic group ownership is intentionally limited to BestEffort.

Implementations

impl KafkaSource

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

Creates a new Kafka source connector with explicit schema.

pub fn with_schema_registry( schema: SchemaRef, config: KafkaSourceConfig, sr_client: SchemaRegistryClient, ) -> Self

Creates a new Kafka source connector with Schema Registry.

pub fn state(&self) -> ConnectorState

Lifecycle state (Created → Initializing → Running → Closed).

pub fn channel_len(&self) -> Arc<AtomicUsize>

Shared backpressure fill counter for downstream wiring.

pub fn rebalance_state(&self) -> Arc<Mutex<RebalanceState>>

Shared partition assignment state (updated by rebalance callbacks).

pub fn has_schema_registry(&self) -> bool

Whether a Schema Registry client is configured.

Trait Implementations

impl Debug for KafkaSource

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for KafkaSource

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl SourceConnector for KafkaSource

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

Observe detached tasks whose lifetime may outlast this connector value.

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

Declare recovery and placement semantics for this exact configuration.

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.

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

Stop advancing external input for an exact cluster transition.

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

Whether the exact provider FIFO boundary has been consumed.

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.

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.

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.

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.

fn schema(&self) -> SchemaRef

Arrow schema of records this source produces.

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.

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 checkpoint(&self) -> SourceCheckpoint

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

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.

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

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

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.

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 cancellation_policy(&self) -> ConnectorCancellationPolicy

Deadline behavior required by the underlying client implementation.

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

Return connector-owned semantic options for durable recovery identity.