Package com.linkedin.davinci.kafka.consumer

Class StoreIngestionTask

java.lang.Object
com.linkedin.davinci.kafka.consumer.StoreIngestionTask
All Implemented Interfaces:
Closeable, AutoCloseable, Runnable
Direct Known Subclasses:
LeaderFollowerStoreIngestionTask
public abstract class StoreIngestionTask extends Object implements Runnable, Closeable
A runnable Kafka Consumer consuming messages from all the partition assigned to current node for a Kafka Topic.

  • Field Details

    • SCHEMA_POLLING_DELAY_MS

      public static long SCHEMA_POLLING_DELAY_MS

    • STORE_VERSION_POLLING_DELAY_MS

      public static long STORE_VERSION_POLLING_DELAY_MS

    • WAITING_TIME_FOR_LAST_RECORD_TO_BE_PROCESSED

      protected static final long WAITING_TIME_FOR_LAST_RECORD_TO_BE_PROCESSED

    • KILL_WAIT_TIME_MS

      protected static final long KILL_WAIT_TIME_MS
      See Also:

    • CHUNK_SCHEMA_ID

      protected static final int CHUNK_SCHEMA_ID

    • REDUNDANT_LOGGING_FILTER

      protected static final RedundantExceptionFilter REDUNDANT_LOGGING_FILTER

    • storageService

      protected final StorageService storageService
      storage destination for consumption

    • storageEngine

      protected final StorageEngine storageEngine

    • kafkaVersionTopic

      protected final String kafkaVersionTopic
      Topics used for this topic consumption TODO: Using a PubSubVersionTopic and PubSubRealTimeTopic extending PubSubTopic for type safety.

    • versionTopic

      protected final PubSubTopic versionTopic

    • realTimeTopic

      protected final PubSubTopic realTimeTopic

    • separateRealTimeTopic

      protected final PubSubTopic separateRealTimeTopic

    • storeName

      protected final String storeName

    • storeVersionName

      protected final String storeVersionName

    • isSystemStore

      protected final boolean isSystemStore

    • versionNumber

      protected final int versionNumber

    • schemaRepository

      protected final ReadOnlySchemaRepository schemaRepository

    • storeRepository

      protected final ReadOnlyStoreRepository storeRepository

    • ingestionTaskName

      protected final String ingestionTaskName

    • kafkaProps

      protected final Properties kafkaProps

    • isRunning

      protected final AtomicBoolean isRunning

    • emitMetrics

      protected final AtomicBoolean emitMetrics

    • consumerActionSequenceNumber

      protected final AtomicInteger consumerActionSequenceNumber

    • consumerActionsQueue

      protected final PriorityBlockingQueue<ConsumerAction> consumerActionsQueue

    • partitionToPendingConsumerActionCountMap

      protected final Map<Integer,AtomicInteger> partitionToPendingConsumerActionCountMap

    • storageMetadataService

      protected final StorageMetadataService storageMetadataService

    • topicManagerRepository

      protected final TopicManagerRepository topicManagerRepository

    • partitionConsumptionStateMap

      protected final ConcurrentMap<Integer,PartitionConsumptionState> partitionConsumptionStateMap
      Per-partition consumption state map

    • storeBufferService

      protected final AbstractStoreBufferService storeBufferService

    • consumerDiv

      protected final DataIntegrityValidator consumerDiv
      The consumer and drainer DIV must remain separate. Since the consumer is always ahead of the drainer, the consumer would be validating data ahead of the actual persisted data on the drainer. NOTE: Currently, the state clearing happens only in drainerDiv which persists its state to disk. consumerDiv is transient, not persisted to disk, and its state is not expected to grow as large. Thus, bouncing effectively clears it (which is not the case for drainerDiv). Later on, we could trigger state cleaning for this consumer DIV as well, if deemed necessary. NOTE: When isGlobalRtDivEnabled() is enabled, this will be used by leaders to produce Global RT DIV state to local VT. This will also be used to send DIV snapshots to the drainer to persist the VT + RT DIV on-disk.

    • consumedBytesSinceLastSync

      protected final VeniceConcurrentHashMap<String,Long> consumedBytesSinceLastSync
      Map of broker URL to the total bytes consumed by ConsumptionTask since the last Global RT DIV sync

    • hostLevelIngestionStats

      protected final HostLevelIngestionStats hostLevelIngestionStats

    • versionedDIVStats

      protected final AggVersionedDIVStats versionedDIVStats

    • versionedIngestionStats

      protected final AggVersionedIngestionStats versionedIngestionStats

    • recordTransformerStats

      protected AggVersionedDaVinciRecordTransformerStats recordTransformerStats

    • isCurrentVersion

      protected final BooleanSupplier isCurrentVersion

    • hybridStoreConfig

      protected final Optional<HybridStoreConfig> hybridStoreConfig

    • divErrorMetricCallback

      protected final Consumer<DataValidationException> divErrorMetricCallback

    • readCycleDelayMs

      protected final long readCycleDelayMs

    • emptyPollSleepMs

      protected final long emptyPollSleepMs

    • diskUsage

      protected final DiskUsage diskUsage

    • databaseSyncBytesIntervalForTransactionalMode

      protected final long databaseSyncBytesIntervalForTransactionalMode
      Message bytes consuming interval before persisting offset in offset db for transactional mode database.

    • databaseSyncBytesIntervalForDeferredWriteMode

      protected final long databaseSyncBytesIntervalForDeferredWriteMode
      Message bytes consuming interval before persisting offset in offset db for deferred-write database.

    • serverConfig

      protected final VeniceServerConfig serverConfig

    • errorPartitionId

      protected final int errorPartitionId
      Used for reporting error when the partitionConsumptionStateMap is empty

    • defaultReadyToServeChecker

      protected final com.linkedin.davinci.kafka.consumer.StoreIngestionTask.ReadyToServeCheck defaultReadyToServeChecker

    • availableSchemaIds

      protected final SparseConcurrentList<Object> availableSchemaIds

    • deserializedSchemaIds

      protected final SparseConcurrentList<Object> deserializedSchemaIds

    • idleCounter

      protected AtomicInteger idleCounter

    • aggKafkaConsumerService

      protected final AggKafkaConsumerService aggKafkaConsumerService

    • writeComputeFailureCode

      protected int writeComputeFailureCode

    • isWriteComputationEnabled

      protected final boolean isWriteComputationEnabled

    • partitionCount

      protected final int partitionCount
      This would be the number of partitions in the StorageEngine and in version topics

    • storeVersionPartitionCount

      protected final int storeVersionPartitionCount

    • bootstrapTimeoutInMs

      protected final long bootstrapTimeoutInMs

    • isIsolatedIngestion

      protected final boolean isIsolatedIngestion

    • ingestionNotificationDispatcher

      protected final com.linkedin.davinci.kafka.consumer.IngestionNotificationDispatcher ingestionNotificationDispatcher

    • chunkAssembler

      protected final InMemoryChunkAssembler chunkAssembler

    • localKafkaServer

      protected final String localKafkaServer

    • localKafkaClusterId

      protected final int localKafkaClusterId

    • localKafkaServerSingletonSet

      protected final Set<String> localKafkaServerSingletonSet

    • isDaVinciClient

      protected final boolean isDaVinciClient

    • isDataRecovery

      protected boolean isDataRecovery

    • dataRecoverySourceVersionNumber

      protected int dataRecoverySourceVersionNumber

    • readOnlyForBatchOnlyStoreEnabled

      protected final boolean readOnlyForBatchOnlyStoreEnabled

    • metaStoreWriter

      protected final MetaStoreWriter metaStoreWriter

    • kafkaClusterUrlResolver

      protected final Function<String,String> kafkaClusterUrlResolver

    • resetErrorReplicaEnabled

      protected final boolean resetErrorReplicaEnabled

    • compressionStrategy

      protected final CompressionStrategy compressionStrategy

    • compressorFactory

      protected final StorageEngineBackedCompressorFactory compressorFactory

    • compressor

      protected final Lazy<VeniceCompressor> compressor

    • isChunked

      protected final boolean isChunked

    • isRmdChunked

      protected final boolean isRmdChunked

    • manifestSerializer

      protected final ChunkedValueManifestSerializer manifestSerializer

    • pubSubContext

      protected final PubSubContext pubSubContext

    • pubSubTopicRepository

      protected final PubSubTopicRepository pubSubTopicRepository

    • recordLevelMetricEnabled

      protected final AtomicBoolean recordLevelMetricEnabled

    • isGlobalRtDivEnabled

      protected final boolean isGlobalRtDivEnabled

    • versionRole

      protected volatile PartitionReplicaIngestionContext.VersionRole versionRole

    • workloadType

      protected volatile PartitionReplicaIngestionContext.WorkloadType workloadType

    • batchReportIncPushStatusEnabled

      protected final boolean batchReportIncPushStatusEnabled

    • parallelProcessingThreadPool

      protected final ExecutorService parallelProcessingThreadPool

    • gracefulShutdownLatch

      protected Lazy<CountDownLatch> gracefulShutdownLatch

    • zkHelixAdmin

      protected Lazy<org.apache.helix.manager.zk.ZKHelixAdmin> zkHelixAdmin

    • hostName

      protected final String hostName

  • Constructor Details

  • Method Details

    • getIngestionBatchProcessor

      protected abstract IngestionBatchProcessor getIngestionBatchProcessor()

    • getStorageEngine

      public StorageEngine getStorageEngine()

    • getIngestionTaskName

      public String getIngestionTaskName()

    • getVersionNumber

      public int getVersionNumber()

    • throwIfNotRunning

      protected void throwIfNotRunning()

    • nextSeqNum

      protected int nextSeqNum()
      Apply an unique and increasing sequence number for each consumer action, so if there are multiple consumer actions in the queue and they have the same priority, whichever be added first into the queue will be polled out first from the queue (FIFO).
      Returns:
      an unique and increasing sequence number for a new consumer action.

    • subscribePartition

      public void subscribePartition(PubSubTopicPartition topicPartition, Optional<PubSubPosition> pubSubPosition)

    • subscribePartition

      public void subscribePartition(PubSubTopicPartition topicPartition, boolean isHelixTriggeredAction, Optional<PubSubPosition> pubSubPosition)
      Adds an asynchronous partition subscription request for the task.

    • unSubscribePartition

      public CompletableFuture<Void> unSubscribePartition(PubSubTopicPartition topicPartition)

    • unSubscribePartition

      public CompletableFuture<Void> unSubscribePartition(PubSubTopicPartition topicPartition, boolean isHelixTriggeredAction)
      Adds an asynchronous partition unsubscription request for the task.

    • dropStoragePartitionGracefully

      public CompletableFuture<Void> dropStoragePartitionGracefully(PubSubTopicPartition topicPartition)
      Drops a storage partition gracefully. This is always a Helix triggered action.

    • hasAnySubscription

      public boolean hasAnySubscription()

    • hasAnyPendingSubscription

      public boolean hasAnyPendingSubscription()

    • isIdleOverThreshold

      public boolean isIdleOverThreshold()
      This helper function will check if the ingestion task has been idle for a long time.

    • resetPartitionConsumptionOffset

      public void resetPartitionConsumptionOffset(PubSubTopicPartition topicPartition)
      Adds an asynchronous resetting partition consumption offset request for the task.

    • getStoreName

      public String getStoreName()

    • isUserSystemStore

      public boolean isUserSystemStore()

    • promoteToLeader

      public abstract void promoteToLeader(PubSubTopicPartition topicPartition, LeaderFollowerPartitionStateModel.LeaderSessionIdChecker checker)

    • demoteToStandby

      public abstract void demoteToStandby(PubSubTopicPartition topicPartition, LeaderFollowerPartitionStateModel.LeaderSessionIdChecker checker)

    • hasPendingPartitionIngestionAction

      public boolean hasPendingPartitionIngestionAction(int userPartition)

    • kill

      public void kill()

    • getStoragePartitionConfig

      protected StoragePartitionConfig getStoragePartitionConfig(PartitionConsumptionState partitionConsumptionState)

    • getStoragePartitionConfig

      protected StoragePartitionConfig getStoragePartitionConfig(boolean sorted, PartitionConsumptionState partitionConsumptionState)

    • isHybridFollower

      protected abstract boolean isHybridFollower(PartitionConsumptionState partitionConsumptionState)

    • checkAndLogIfLagIsAcceptableForHybridStore

      protected abstract boolean checkAndLogIfLagIsAcceptableForHybridStore(PartitionConsumptionState partitionConsumptionState, long lag, long threshold, boolean shouldLogLag, LagType lagType)
      Checks whether the lag is acceptable for hybrid stores

    • isReadyToServe

      protected boolean isReadyToServe(PartitionConsumptionState partitionConsumptionState)
      This function checks various conditions to verify if a store is ready to serve. Lag = (Source Max Offset - SOBR Source Offset) - (Current Offset - SOBR Destination Offset)
      Returns:
      true if EOP was received and (for hybrid stores) if lag <= threshold

    • isRealTimeBufferReplayStarted

      protected abstract boolean isRealTimeBufferReplayStarted(PartitionConsumptionState partitionConsumptionState)

    • measureHybridOffsetLag

      protected abstract long measureHybridOffsetLag(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Measure the hybrid offset lag for replica being tracked in `partitionConsumptionState`.

    • measureHybridHeartbeatLag

      protected abstract long measureHybridHeartbeatLag(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Measure the hybrid heartbeat lag for replica being tracked in `partitionConsumptionState`. If it is Da Vinci client, return HeartbeatMonitoringService.INVALID_HEARTBEAT_LAG before it is implemented.

    • measureHybridHeartbeatTimestamp

      protected abstract long measureHybridHeartbeatTimestamp(PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)
      Measure the hybrid heartbeat timestamp for replica being tracked in `partitionConsumptionState`. If it is Da Vinci client, return HeartbeatMonitoringService.INVALID_MESSAGE_TIMESTAMP before it is implemented.

    • reportIfCatchUpVersionTopicOffset

      protected abstract void reportIfCatchUpVersionTopicOffset(PartitionConsumptionState partitionConsumptionState)
      Check if the ingestion progress has reached to the end of the version topic. This is currently only used LeaderFollowerStoreIngestionTask.

    • produceToStoreBufferService

      protected void produceToStoreBufferService(DefaultPubSubMessage consumedRecord, LeaderProducedRecordContext leaderProducedRecordContext, int partition, String kafkaUrl, long beforeProcessingRecordTimestampNs, long currentTimeForMetricsMs) throws InterruptedException
      This function will produce a pair of consumer record and a it's derived produced record to the writer buffers maintained by StoreBufferService.
      Parameters:
      consumedRecord - : received consumer record
      leaderProducedRecordContext - : derived leaderProducedRecordContext
      partition -
      kafkaUrl -
      Throws:
      InterruptedException

    • validateAndFilterOutDuplicateMessagesFromLeaderTopic

      protected abstract Iterable<DefaultPubSubMessage> validateAndFilterOutDuplicateMessagesFromLeaderTopic(Iterable<DefaultPubSubMessage> records, String kafkaUrl, PubSubTopicPartition topicPartition)

    • produceToStoreBufferServiceOrKafka

      protected void produceToStoreBufferServiceOrKafka(Iterable<DefaultPubSubMessage> records, PubSubTopicPartition topicPartition, String kafkaUrl, int kafkaClusterId) throws InterruptedException
      This function is in charge of producing the consumer records to the writer buffers maintained by StoreBufferService. This function may modify the original record in KME and it is unsafe to use the payload from KME directly after this call.
      Parameters:
      records - : received consumer records
      topicPartition -
      Throws:
      InterruptedException

    • produceToStoreBufferServiceOrKafkaInBatch

      protected void produceToStoreBufferServiceOrKafkaInBatch(Iterable<DefaultPubSubMessage> records, PubSubTopicPartition topicPartition, PartitionConsumptionState partitionConsumptionState, String kafkaUrl, int kafkaClusterId) throws InterruptedException
      Throws:
      InterruptedException

    • checkIngestionProgress

      protected void checkIngestionProgress(Store store) throws InterruptedException
      Throws:
      InterruptedException

    • refreshIngestionContextIfChanged

      protected void refreshIngestionContextIfChanged(Store store) throws InterruptedException
      Throws:
      InterruptedException

    • isIngestionTaskActive

      public boolean isIngestionTaskActive()

    • run

      public void run()
      Polls the producer for new messages in an infinite loop by a dedicated consumer thread and processes the new messages by current thread.
      Specified by:
      run in interface Runnable

    • updateOffsetMetadataAndSyncOffset

      protected void updateOffsetMetadataAndSyncOffset(@Nonnull PartitionConsumptionState pcs)

    • updateOffsetMetadataAndSyncOffset

      protected void updateOffsetMetadataAndSyncOffset(DataIntegrityValidator div, @Nonnull PartitionConsumptionState pcs)

    • updateAndSyncOffsetFromSnapshot

      protected void updateAndSyncOffsetFromSnapshot(PartitionTracker vtDivSnapshot, PubSubTopicPartition topicPartition)
      This version of the method syncs using a PartitionTracker object which contains the vtSegments and LCVP

    • closeVeniceWriters

      public void closeVeniceWriters(boolean doFlush)

    • closeVeniceViewWriters

      protected void closeVeniceViewWriters(boolean doFlush)

    • resolveSourceKafkaServersWithinTopicSwitch

      protected TopicSwitch resolveSourceKafkaServersWithinTopicSwitch(TopicSwitch originalTopicSwitch)
      Applies name resolution to all Kafka URLs in the provided TopicSwitch. Useful for translating URLs that came from a different runtime (e.g. from the controller, or from state persisted by a previous run of the same server).
      Returns:
      the same TopicSwitch, mutated such that all Kafka URLs it contains are guaranteed to be usable by the current Venice server instance

    • getOffsetToOnlineLagThresholdPerPartition

      protected static long getOffsetToOnlineLagThresholdPerPartition(Optional<HybridStoreConfig> hybridStoreConfig, String storeName, int partitionCount)

    • processCommonConsumerAction

      protected void processCommonConsumerAction(ConsumerAction consumerAction) throws InterruptedException
      Throws:
      InterruptedException

    • getTopicPartitionEndPosition

      protected PubSubPosition getTopicPartitionEndPosition(String pubSubBrokerAddress, PubSubTopicPartition topicPartition)
      Returns:
      the end position for the topic partition in SIT, or a PubSubSymbolicPosition.LATEST value if it failed to get it. N.B.: The returned end position is the last successfully replicated message plus one. If the partition has never been written to, the end position is equal to the start position.

    • checkLongRunningTaskState

      protected abstract void checkLongRunningTaskState() throws InterruptedException
      Throws:
      InterruptedException

    • processConsumerAction

      protected abstract void processConsumerAction(ConsumerAction message, Store store) throws InterruptedException
      Throws:
      InterruptedException

    • getConsumptionSourceKafkaAddress

      protected abstract Set<String> getConsumptionSourceKafkaAddress(PartitionConsumptionState partitionConsumptionState)

    • startConsumingAsLeader

      protected void startConsumingAsLeader(PartitionConsumptionState partitionConsumptionState)

    • getRealTimeDataSourceKafkaAddress

      protected Set<String> getRealTimeDataSourceKafkaAddress(PartitionConsumptionState partitionConsumptionState)

    • getPartitionConsumptionState

      public PartitionConsumptionState getPartitionConsumptionState(int partitionId)

    • hasAnyPartitionConsumptionState

      public boolean hasAnyPartitionConsumptionState(Predicate<PartitionConsumptionState> pcsPredicate)

    • getFailedIngestionPartitionCount

      public int getFailedIngestionPartitionCount()

    • shouldProcessRecord

      protected boolean shouldProcessRecord(DefaultPubSubMessage record)
      Common record check for different state models: check whether server continues receiving messages after EOP for a batch-only store.

    • shouldPersistRecord

      protected boolean shouldPersistRecord(DefaultPubSubMessage record, PartitionConsumptionState partitionConsumptionState)

    • processConsumerRecord

      public void processConsumerRecord(DefaultPubSubMessage record, LeaderProducedRecordContext leaderProducedRecordContext, int partition, String kafkaUrl, long beforeProcessingRecordTimestampNs)
      This function will be invoked in StoreBufferService to process buffered PubSubMessage.

    • recordHeartbeatReceived

      protected void recordHeartbeatReceived(PartitionConsumptionState partitionConsumptionState, DefaultPubSubMessage consumerRecord, String kafkaUrl)

    • setLastConsumerException

      public void setLastConsumerException(Exception e)

    • setLastStoreIngestionException

      public void setLastStoreIngestionException(Exception e)

    • recordChecksumVerificationFailure

      public void recordChecksumVerificationFailure()

    • recordAssembledRecordSize

      protected void recordAssembledRecordSize(int keyLen, ByteBuffer valueBytes, ByteBuffer rmdBytes, long currentTimeMs)
      Records metrics for the original size of full-assembled records (key + value) and RMD by utilizing the field ChunkedValueManifest.size. Also records the ratio of assembled record size to maximum allowed size, which is intended to be used to alert customers about how close they are to hitting the size limit.
      Parameters:
      keyLen - The size of the record's key
      valueBytes - Put.putValue which is expected to be a serialized ChunkedValueManifest
      rmdBytes - Put.replicationMetadataPayload which can be a serialized ChunkedValueManifest if RMD chunking was enabled or just the RMD payload otherwise

    • recordAssembledRecordSizeRatio

      protected abstract void recordAssembledRecordSizeRatio(double ratio, long currentTimeMs)

    • calculateAssembledRecordSizeRatio

      protected abstract double calculateAssembledRecordSizeRatio(long recordSize)

    • measureLagWithCallToPubSub

      protected long measureLagWithCallToPubSub(String pubSubServerName, PubSubTopicPartition pubSubTopicPartition, PubSubPosition currentPosition)
      Parameters:
      pubSubServerName - Pub Sub deployment to interrogate
      pubSubTopicPartition - the topic partition to measure lag for
      Returns:
      the lag, or 9223372036854775807L if it failed to measure it N.B.: Note that the returned lag can be negative since the end offset used in the calculation is cached.

    • measureLagWithCallToPubSub

      protected static long measureLagWithCallToPubSub(String pubSubBrokerAddress, PubSubTopicPartition pubSubTopicPartition, PubSubPosition currentPosition, Function<String,TopicManager> topicManagerProvider)

    • getWriteComputeErrorCode

      public abstract int getWriteComputeErrorCode()

    • updateLeaderTopicOnFollower

      public abstract void updateLeaderTopicOnFollower(PartitionConsumptionState partitionConsumptionState)

    • minZeroLag

      protected long minZeroLag(long value)
      Because of timing considerations, it is possible that some lag metrics could compute negative values. Negative lag does not make sense so the intent is to ease interpretation by applying a lower bound of zero on these metrics...

    • isHybridMode

      public boolean isHybridMode()

    • processEndOfPush

      protected void processEndOfPush(KafkaMessageEnvelope endOfPushKME, PubSubPosition offset, PartitionConsumptionState partitionConsumptionState, EndOfPush endOfPush)

    • processStartOfIncrementalPush

      protected void processStartOfIncrementalPush(ControlMessage startOfIncrementalPush, PartitionConsumptionState partitionConsumptionState, long pubSubMessageTime)

    • processEndOfIncrementalPush

      protected void processEndOfIncrementalPush(ControlMessage endOfIncrementalPush, PartitionConsumptionState partitionConsumptionState, long pubSubMessageTime)

    • processControlMessageForViews

      protected void processControlMessageForViews(KafkaKey kafkaKey, KafkaMessageEnvelope kafkaMessageEnvelope, ControlMessage controlMessage, int partition, PartitionConsumptionState partitionConsumptionState)
      This isn't used for ingestion outside L/F, so we NoOp here and rely on the actual implementation in LeaderFollowerStoreIngestionTask

    • processTopicSwitch

      protected void processTopicSwitch(ControlMessage controlMessage, int partition, PubSubPosition offset, PartitionConsumptionState partitionConsumptionState)

    • updateOffsetMetadataInOffsetRecord

      protected abstract void updateOffsetMetadataInOffsetRecord(PartitionConsumptionState partitionConsumptionState)
      Sync the metadata about offset in OffsetRecord. PartitionConsumptionState will pass through some information to OffsetRecord for persistence and Offset rewind/split brain has been guarded in updateLatestInMemoryProcessedOffset(com.linkedin.davinci.kafka.consumer.PartitionConsumptionState, com.linkedin.venice.pubsub.api.DefaultPubSubMessage, com.linkedin.davinci.kafka.consumer.LeaderProducedRecordContext, java.lang.String, boolean).

    • updateLatestInMemoryProcessedOffset

      protected abstract void updateLatestInMemoryProcessedOffset(PartitionConsumptionState partitionConsumptionState, DefaultPubSubMessage consumerRecordWrapper, LeaderProducedRecordContext leaderProducedRecordContext, String kafkaUrl, boolean dryRun)
      Maintain the latest processed offsets by drainers in memory; in most of the time, these offsets are ahead of the checkpoint offsets inside OffsetRecord. Prior to update the offset in memory, the underlying storage engine should have persisted the given record. Dry-run mode will only do offset rewind check and it won't update the processed offset.

    • recordWriterStats

      protected abstract void recordWriterStats(long consumerTimestampMs, long producerBrokerLatencyMs, long brokerConsumerLatencyMs, PartitionConsumptionState partitionConsumptionState)

    • validateMessage

      protected void validateMessage(PartitionTracker.TopicType type, DataIntegrityValidator validator, DefaultPubSubMessage consumerRecord, PartitionConsumptionState partitionConsumptionState, boolean tolerateMissingMessagesForRealTimeTopic)
      Message validation using DIV. Leaders should pass in the validator instance from LeaderFollowerStoreIngestionTask; and drainers should pass in the validator instance from StoreIngestionTask 1. If valid DIV errors happen after EOP is received, no fatal exceptions will be thrown. But the errors will be recorded into the DIV metrics. 2. For any DIV errors happened to unregistered producers && after EOP, the errors will be ignored. 3. For any DIV errors happened to records which is after logCompactionDelayInMs, the errors will be ignored.

    • cloneDrainerDivProducerStates

      protected void cloneDrainerDivProducerStates(int partition, DataIntegrityValidator validator)
      We should only allow StoreIngestionTask to access drainerDiv; other components like leaders in LeaderFollowerStoreIngestionTask should never access the DIV validator in drainer, because messages consumption in leader is ahead of drainer, leaders and drainers are processing messages at different paces.

    • putInStorageEngine

      protected void putInStorageEngine(int partition, byte[] keyBytes, Put put)
      Persist Put record to storage engine.

    • removeFromStorageEngine

      protected void removeFromStorageEngine(int partition, byte[] keyBytes, Delete delete)

    • throwOrLogStorageFailureDependingIfStillSubscribed

      protected void throwOrLogStorageFailureDependingIfStillSubscribed(int partition, VeniceException e)

    • logStorageOperationWhileUnsubscribed

      protected void logStorageOperationWhileUnsubscribed(int partition)

    • consumerHasAnySubscription

      public boolean consumerHasAnySubscription()

    • consumerHasSubscription

      public boolean consumerHasSubscription(PubSubTopic topic, PartitionConsumptionState partitionConsumptionState)

    • unsubscribeFromTopic

      protected void unsubscribeFromTopic(PubSubTopic topic, PartitionConsumptionState partitionConsumptionState)
      This method unsubscribes topic-partition from the input. If it is real-time topic and separate RT topic is enabled, it will also unsubscribe from separate real-time topic.

    • consumerBatchUnsubscribe

      public void consumerBatchUnsubscribe(Set<PubSubTopicPartition> topicPartitionSet)

    • consumerUnSubscribeAllTopics

      public abstract void consumerUnSubscribeAllTopics(PartitionConsumptionState partitionConsumptionState)

    • consumerSubscribe

      public void consumerSubscribe(PubSubTopic pubSubTopic, PartitionConsumptionState partitionConsumptionState, PubSubPosition startOffset, String kafkaURL)
      This method will try to resolve actual topic-partition from input Kafka URL and subscribe to the resolved topic-partition.

    • consumerResetOffset

      public void consumerResetOffset(PubSubTopic topic, PartitionConsumptionState partitionConsumptionState)

    • waitVersionStateAvailable

      protected StoreVersionState waitVersionStateAvailable(String kafkaTopic) throws InterruptedException
      Throws:
      InterruptedException

    • close

      public void close()
      Stops the consumer task.
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Closeable

    • shutdownAndWait

      public boolean shutdownAndWait(int waitTime)
      This method is a blocking call to wait for StoreIngestionTask for fully shutdown in the given time.
      Parameters:
      waitTime - Maximum wait time for the shutdown operation.
      Returns:
      whether able to gracefully shut down within the waitTime

    • isRunning

      public boolean isRunning()
      A function to allow the service to get the current status of the task. This would allow the service to create a new task if required.

    • getVersionTopic

      public PubSubTopic getVersionTopic()

    • isMetricsEmissionEnabled

      public boolean isMetricsEmissionEnabled()

    • enableMetricsEmission

      public void enableMetricsEmission()

    • disableMetricsEmission

      public void disableMetricsEmission()

    • isPartitionConsumingOrHasPendingIngestionAction

      public boolean isPartitionConsumingOrHasPendingIngestionAction(int userPartition)
      To check whether the given partition is still consuming message from Kafka

    • createKafkaConsumerProperties

      protected Properties createKafkaConsumerProperties(Properties localConsumerProps, String remoteKafkaSourceAddress, boolean consumeRemotely)
      Override the ConfigKeys.KAFKA_BOOTSTRAP_SERVERS config with a remote Kafka bootstrap url.

    • resubscribe

      protected abstract void resubscribe(PartitionConsumptionState partitionConsumptionState) throws InterruptedException
      Throws:
      InterruptedException

    • reportError

      public void reportError(String message, int userPartition, Exception e)

    • isActiveActiveReplicationEnabled

      public boolean isActiveActiveReplicationEnabled()

    • dumpPartitionConsumptionStates

      public void dumpPartitionConsumptionStates(AdminResponse response, ComplementSet<Integer> partitions)
      Invoked by admin request to dump the requested partition consumption states

    • dumpStoreVersionState

      public void dumpStoreVersionState(AdminResponse response)
      Invoked by admin request to dump store version state metadata.

    • getServerConfig

      public VeniceServerConfig getServerConfig()

    • updateOffsetMetadataAndSync

      public void updateOffsetMetadataAndSync(int partitionId)

    • getTopicManager

      protected TopicManager getTopicManager(String sourceKafkaServer)
      The function returns local or remote topic manager.
      Parameters:
      sourceKafkaServer - The address of source kafka bootstrap server.
      Returns:
      topic manager

    • waitForAllMessageToBeProcessedFromTopicPartition

      protected void waitForAllMessageToBeProcessedFromTopicPartition(PubSubTopicPartition topicPartition, PartitionConsumptionState partitionConsumptionState) throws InterruptedException
      The purpose of this function is to wait for the complete processing (including persistence to disk) of all the messages those were consumed from this kafka {topic, partition} prior to calling this function. This will make the calling thread to block.
      Parameters:
      topicPartition - for which to wait
      Throws:
      InterruptedException

    • delegateConsumerRecord

      protected abstract StoreIngestionTask.DelegateConsumerRecordResult delegateConsumerRecord(PubSubMessageProcessedResultWrapper consumerRecordWrapper, int partition, String kafkaUrl, int kafkaClusterId, long beforeProcessingPerRecordTimestampNs, long beforeProcessingBatchRecordsTimestampMs)

    • recordProcessedRecordStats

      protected void recordProcessedRecordStats(PartitionConsumptionState partitionConsumptionState, int processedRecordSize)

    • isTransientRecordBufferUsed

      public boolean isTransientRecordBufferUsed(PartitionConsumptionState partitionConsumptionState)
      This is not a per record state. Rather it's used to indicate if the transient record buffer is being used at all for this ingestion task/partition or not. The criterias are the following: 1. For L/F mode only WC ingestion task needs this buffer. 2. The transient record buffer is only used post-EOP.

    • setPartitionConsumptionState

      protected void setPartitionConsumptionState(int partition, PartitionConsumptionState pcs)

    • getVersionedDIVStats

      protected AggVersionedDIVStats getVersionedDIVStats()

    • getVersionIngestionStats

      protected AggVersionedIngestionStats getVersionIngestionStats()

    • getCompressionStrategy

      protected CompressionStrategy getCompressionStrategy()

    • getCompressor

      protected Lazy<VeniceCompressor> getCompressor()

    • isChunked

      protected boolean isChunked()

    • getSchemaRepo

      protected ReadOnlySchemaRepository getSchemaRepo()

    • getHostLevelIngestionStats

      protected HostLevelIngestionStats getHostLevelIngestionStats()

    • getKafkaVersionTopic

      protected String getKafkaVersionTopic()

    • extractUpstreamPosition

      protected PubSubPosition extractUpstreamPosition(DefaultPubSubMessage consumerRecord)
      Extract the upstream position from the given consumer record's leader metadata.
      Parameters:
      consumerRecord - the consumer record to extract upstream position from
      Returns:
      the upstream position if available, otherwise PubSubSymbolicPosition.EARLIEST

    • extractUpstreamClusterId

      protected int extractUpstreamClusterId(DefaultPubSubMessage consumerRecord)

    • maybeSendIngestionHeartbeat

      protected abstract Set<String> maybeSendIngestionHeartbeat()
      For L/F hybrid stores, the leader periodically writes a special SOS message to the RT topic. Check LeaderFollowerStoreIngestionTask.maybeSendIngestionHeartbeat() for more details.

    • isProducingVersionTopicHealthy

      public boolean isProducingVersionTopicHealthy()
      This function is checking the following conditions: 1. Whether the version topic exists or not.

    • isCurrentVersion

      public boolean isCurrentVersion()

    • hasAllPartitionReportedCompleted

      public boolean hasAllPartitionReportedCompleted()

    • isSeparatedRealtimeTopicEnabled

      public boolean isSeparatedRealtimeTopicEnabled()

    • getDataIntegrityValidator

      public DataIntegrityValidator getDataIntegrityValidator()
      When Global RT DIV is enabled the ConsumptionTask's DIV is exclusively used to validate data integrity.

    • getStorageUtilizationManager

      public StorageUtilizationManager getStorageUtilizationManager()

    • hasReplicas

      protected boolean hasReplicas()
      Checks whether there are any replicas currently present on this host.

      For batch-only stores, a replica might be removed from the PCS map but still physically present on the host. activeReplicaCount counter tracks the replica's lifecycle from the bootstrap stage through to the drop stage.

    • validateEndOfPushReceivedBeforeTopicSwitch

      protected static void validateEndOfPushReceivedBeforeTopicSwitch(PartitionConsumptionState partitionConsumptionState, PubSubPosition position)
      Validates that END_OF_PUSH has been received before processing TOPIC_SWITCH.
      Parameters:
      partitionConsumptionState - The partition consumption state to validate
      position - The position/offset for error reporting
      Throws:
      VeniceException - if END_OF_PUSH has not been received