Package com.linkedin.davinci.kafka.consumer

Class IngestionBatchProcessor

java.lang.Object

com.linkedin.davinci.kafka.consumer.IngestionBatchProcessor

public class IngestionBatchProcessor
extends Object

This class is used to process the AA/WC messages in parallel to address the long-tail partition lagging issues. For the AA/WC message handling, the consumption is not the bottleneck, but the processing overhead, and even with a single consumer, with IngestionBatchProcessor, we hope we can utilize the full node's resources to speed up the leader ingestion.

Constructor Summary

Constructors

Constructor

Description

IngestionBatchProcessor(String storeVersionName, ExecutorService batchProcessingThreadPool, KeyLevelLocksManager lockManager, com.linkedin.davinci.kafka.consumer.IngestionBatchProcessor.ProcessingFunction processingFunction, boolean isWriteComputationEnabled, boolean isActiveActiveReplicationEnabled, AggVersionedIngestionStats aggVersionedIngestionStats, HostLevelIngestionStats hostLevelIngestionStats)

Method Summary

Modifier and Type	Method	Description
static boolean	isAllMessagesFromRTTopic(Iterable<PubSubMessage<KafkaKey,KafkaMessageEnvelope,Long>> records)
NavigableMap<ByteArrayKey,ReentrantLock>	lockKeys(List<PubSubMessage<KafkaKey,KafkaMessageEnvelope,Long>> records)	When lockManager is not null, this function will try to lock all the keys (except Control Messages) passed by the params.
List<PubSubMessageProcessedResultWrapper<KafkaKey,KafkaMessageEnvelope,Long>>	process(List<PubSubMessage<KafkaKey,KafkaMessageEnvelope,Long>> records, PartitionConsumptionState partitionConsumptionState, int partition, String kafkaUrl, int kafkaClusterId, long beforeProcessingRecordTimestampNs, long beforeProcessingBatchRecordsTimestampMs)
void	unlockKeys(NavigableMap<ByteArrayKey,ReentrantLock> keyLockMap)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

IngestionBatchProcessor

public IngestionBatchProcessor(String storeVersionName,
 ExecutorService batchProcessingThreadPool,
 KeyLevelLocksManager lockManager,
 com.linkedin.davinci.kafka.consumer.IngestionBatchProcessor.ProcessingFunction processingFunction,
 boolean isWriteComputationEnabled,
 boolean isActiveActiveReplicationEnabled,
 AggVersionedIngestionStats aggVersionedIngestionStats,
 HostLevelIngestionStats hostLevelIngestionStats)

Method Details

lockKeys

public NavigableMap<ByteArrayKey,ReentrantLock> lockKeys(List<PubSubMessage<KafkaKey,KafkaMessageEnvelope,Long>> records)

When lockManager is not null, this function will try to lock all the keys (except Control Messages) passed by the params.

unlockKeys

public void unlockKeys(NavigableMap<ByteArrayKey,ReentrantLock> keyLockMap)

isAllMessagesFromRTTopic

public static boolean isAllMessagesFromRTTopic(Iterable<PubSubMessage<KafkaKey,KafkaMessageEnvelope,Long>> records)

process

public List<PubSubMessageProcessedResultWrapper<KafkaKey,KafkaMessageEnvelope,Long>> process(List<PubSubMessage<KafkaKey,KafkaMessageEnvelope,Long>> records,
 PartitionConsumptionState partitionConsumptionState,
 int partition,
 String kafkaUrl,
 int kafkaClusterId,
 long beforeProcessingRecordTimestampNs,
 long beforeProcessingBatchRecordsTimestampMs)