Struct KafkaSink 

pub struct KafkaSink { /* private fields */ }

Kafka sink connector that writes Arrow RecordBatch data to Kafka topics.

Operates in Ring 1 (background) receiving data from Ring 0 via the subscription API.

Lifecycle

1. Create with KafkaSink::new
2. Call open() to create the producer and connect to Kafka
3. For each epoch:
   • begin_epoch() starts a Kafka transaction (exactly-once only)
   • write_batch() serializes and produces records
   • commit_epoch() commits the transaction
4. Call close() for clean shutdown

Implementations

impl KafkaSink

pub fn new(schema: SchemaRef, config: KafkaSinkConfig) -> Self

Creates a new Kafka sink connector with explicit schema.

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

Creates a new Kafka sink with Schema Registry integration.

pub fn state(&self) -> ConnectorState

Lifecycle state (Created → Running → Closed).

pub fn has_schema_registry(&self) -> bool

Whether Avro schema registration is available.

pub fn current_epoch(&self) -> u64

Active epoch (incremented by checkpoint coordinator).

pub fn last_committed_epoch(&self) -> u64

Last epoch that was successfully committed to Kafka.

Trait Implementations

impl Debug for KafkaSink

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

Formats the value using the given formatter.

impl SinkConnector for KafkaSink

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,

Opens the connector and initializes the connection.

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,

Writes a batch of records to the external system.

fn schema(&self) -> SchemaRef

Returns the expected input schema for this sink.

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,

Begins a new epoch for exactly-once processing.

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

Pre-commits the current epoch (phase 1 of two-phase commit).

fn commit_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,

Commits the current epoch (phase 2 of two-phase commit).

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,

Rolls back the current epoch.

fn health_check(&self) -> HealthStatus

Returns the current health status of the connector.

fn metrics(&self) -> ConnectorMetrics

Returns current metrics from the connector.

fn capabilities(&self) -> SinkConnectorCapabilities

Returns the capabilities of this sink connector.

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,

Flushes any buffered data to the external system.

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,

Closes the connector and releases all resources.

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

Returns this connector as a SchemaRegistryAware, if supported.

fn as_schema_evolvable(&self) -> Option<&dyn SchemaEvolvable>

Returns this connector as a SchemaEvolvable, if supported.