Struct mz_storage::sink::kafka::TransactionalProducer

struct TransactionalProducer {

    task_name: String,
    data_topic: String,
    progress_topic: String,
    progress_key: ProgressKey,
    sink_version: u64,
    partition_strategy: SinkPartitionStrategy,
    partition_count: Arc<AtomicU64>,
    _partition_count_task: AbortOnDropHandle<()>,
    producer: ThreadedProducer<TunnelingClientContext<MzClientContext>>,
    statistics: SinkStatistics,
    staged_messages: u64,
    staged_bytes: u64,
    socket_timeout: Duration,
    transaction_timeout: Duration,
}

Fields

task_name: String

The task name used for any blocking calls spawned onto the tokio threadpool.

data_topic: String

The topic where all the updates go.

progress_topic: String

The topic where all the upper frontiers go.

progress_key: ProgressKey

The key each progress record is associated with.

sink_version: u64

The version of this sink, used to fence out previous versions from writing.

partition_strategy: SinkPartitionStrategy

The strategy to partition the data with.

partition_count: Arc<AtomicU64>

The number of partitions in the target topic.

_partition_count_task: AbortOnDropHandle<()>

A task to periodically refresh the partition count.

producer: ThreadedProducer<TunnelingClientContext<MzClientContext>>

The underlying Kafka producer.

statistics: SinkStatistics

A handle to the metrics associated with this sink.

staged_messages: u64

The number of messages staged for the currently open transactions. It is reset to zero every time a transaction commits.

staged_bytes: u64

The total number bytes staged for the currently open transactions. It is reset to zero every time a transaction commits.

socket_timeout: Duration

The timeout to use for network operations.

transaction_timeout: Duration

The timeout to use for committing transactions.

Implementations

impl TransactionalProducer

async fn new( sink_id: GlobalId, connection: &KafkaSinkConnection, partition_strategy: SinkPartitionStrategy, storage_configuration: &StorageConfiguration, metrics: Arc<KafkaSinkMetrics>, statistics: SinkStatistics, sink_version: u64, ) -> Result<(Self, Antichain<Timestamp>), ContextCreationError>

Initializes a transcational producer for the sink identified by sink_id. After this call returns it is guranteed that all previous TransactionalProducer instances for the same sink have been fenced out (i.e init_transations() has been called successfully).

async fn spawn_blocking<F, R>(&self, f: F) -> Result<R, ContextCreationError>

where F: FnOnce(ThreadedProducer<TunnelingClientContext<MzClientContext>>) -> Result<R, KafkaError> + Send + 'static, R: Send + 'static,

Runs the blocking operation f on the producer in the tokio threadpool and checks for SSH status in case of failure.

async fn begin_transaction(&self) -> Result<(), ContextCreationError>

fn send( &mut self, message: &KafkaMessage, time: Timestamp, diff: Diff, ) -> Result<(), KafkaError>

Synchronously puts the provided message to librdkafka’s send queue. This method only returns an error if the queue is full. Handling this error by buffering the message and retrying is equivalent to adjusting the maximum number of queued items in rdkafka so it is adviced that callers only handle this error in order to apply backpressure to the rest of the system.

async fn commit_transaction( &mut self, upper: Antichain<Timestamp>, ) -> Result<(), ContextCreationError>

Commits all the staged updates of the currently open transaction plus a progress record describing upper to the progress topic.