Struct BackgroundProducerWorker

pub(crate) struct BackgroundProducerWorker {
    pub(crate) provider: Arc<KafkaProvider>,
    pub(crate) client: Arc<dyn ProducerClient>,
    pub(crate) compression: Compression,
    pub(crate) receiver: Receiver<WorkerRequest>,
    pub(crate) request_batch_size: usize,
    pub(crate) max_batch_bytes: usize,
    pub(crate) index_collector: Box<dyn IndexCollector>,
    pub(crate) high_watermark: Arc<DashMap<Arc<KafkaProvider>, u64>>,
}

Fields

provider: Arc<KafkaProvider>

client: Arc<dyn ProducerClient>

The ProducerClient.

compression: Compression

receiver: Receiver<WorkerRequest>

Receiver of ProduceRequest.

request_batch_size: usize

Max batch size for a worker to handle requests.

max_batch_bytes: usize

Max bytes size for a single flush.

index_collector: Box<dyn IndexCollector>

Collecting ids of WAL entries.

high_watermark: Arc<DashMap<Arc<KafkaProvider>, u64>>

High watermark for all topics.

Implementations

impl BackgroundProducerWorker

pub(crate) async fn dump_index(&mut self, req: DumpIndexRequest)

impl BackgroundProducerWorker

async fn do_flush(&mut self, __arg1: PendingRequest)

pub(crate) async fn try_flush_pending_requests( &mut self, pending_requests: Vec<PendingRequest>, )

impl BackgroundProducerWorker

pub(crate) fn aggregate_records( &self, requests: &mut Vec<ProduceRequest>, max_batch_bytes: usize, ) -> Vec<PendingRequest>

Aggregates records into batches, ensuring that the size of each batch does not exceed a specified maximum (max_batch_bytes).

Panic

Panic if any [Record]’s approximate_size > max_batch_bytes.

impl BackgroundProducerWorker

pub async fn update_high_watermark(&mut self)

Updates the high watermark for the topic.

This function retrieves the latest offset from Kafka and updates the high watermark in the shared map.

impl BackgroundProducerWorker

pub(crate) async fn run(&mut self)

async fn handle_requests(&mut self, buffer: &mut Vec<WorkerRequest>)