Struct mz_storage::source::kafka::KafkaSourceReader

pub struct KafkaSourceReader {

    topic_name: String,
    source_name: String,
    id: GlobalId,
    consumer: Arc<BaseConsumer<TunnelingClientContext<GlueConsumerContext>>>,
    partition_consumers: Vec<PartitionConsumer>,
    worker_id: usize,
    worker_count: usize,
    last_offsets: BTreeMap<usize, BTreeMap<PartitionId, i64>>,
    start_offsets: BTreeMap<PartitionId, i64>,
    stats_rx: Receiver<Jsonb>,
    progress_statistics: Arc<Mutex<PartialProgressStatistics>>,
    partition_metrics: KafkaSourceMetrics,
    partition_capabilities: BTreeMap<PartitionId, PartitionCapability>,
}

Contains all information necessary to ingest data from Kafka

Fields

topic_name: String

Name of the topic on which this source is backed on

source_name: String

Name of the source (will have format kafka-source-id)

id: GlobalId

Source global ID

consumer: Arc<BaseConsumer<TunnelingClientContext<GlueConsumerContext>>>

Kafka consumer for this source

partition_consumers: Vec<PartitionConsumer>

List of consumers. A consumer should be assigned per partition to guarantee fairness

worker_id: usize

Worker ID

worker_count: usize

Total count of workers

last_offsets: BTreeMap<usize, BTreeMap<PartitionId, i64>>

The most recently read offset for each partition known to this source reader by output-index. An offset of -1 indicates that no prior message has been read for the given partition.

start_offsets: BTreeMap<PartitionId, i64>

The offset to start reading from for each partition.

stats_rx: Receiver<Jsonb>

Channel to receive Kafka statistics JSON blobs from the stats callback.

progress_statistics: Arc<Mutex<PartialProgressStatistics>>

Progress statistics as collected from the resume_uppers stream and the partition metadata thread.

partition_metrics: KafkaSourceMetrics

A handle to the partition specific metrics

partition_capabilities: BTreeMap<PartitionId, PartitionCapability>

Per partition capabilities used to produce messages

Implementations

impl KafkaSourceReader

fn ensure_partition(&mut self, pid: PartitionId)

Ensures that a partition queue for pid exists.

fn create_partition_queue( &mut self, partition_id: PartitionId, initial_offset: Offset, )

Creates a new partition queue for partition_id.

fn update_stats(&mut self)

Read any statistics JSON blobs generated via the rdkafka statistics callback.

fn handle_message( &mut self, message: Result<SourceMessage, KafkaHeaderParseError>, (partition, offset): (PartitionId, MzOffset), output_index: &usize, ) -> Option<(Result<SourceMessage, KafkaHeaderParseError>, KafkaTimestamp, Diff)>

Checks if the given message is viable for emission. This checks if the message offset is past the expected offset and returns None if it is not.