Package com.linkedin.davinci.kafka.consumer

Class ActiveActiveStoreIngestionTask

java.lang.Object

com.linkedin.davinci.kafka.consumer.StoreIngestionTask

com.linkedin.davinci.kafka.consumer.LeaderFollowerStoreIngestionTask

com.linkedin.davinci.kafka.consumer.ActiveActiveStoreIngestionTask

All Implemented Interfaces:

java.io.Closeable, java.lang.AutoCloseable, java.lang.Runnable

public class ActiveActiveStoreIngestionTask
extends LeaderFollowerStoreIngestionTask

This class contains logic that SNs must perform if a store-version is running in Active/Active mode.

Nested Class Summary

Nested classes/interfaces inherited from class com.linkedin.davinci.kafka.consumer.StoreIngestionTask

StoreIngestionTask.DelegateConsumerRecordResult

Field Summary

Fields inherited from class com.linkedin.davinci.kafka.consumer.LeaderFollowerStoreIngestionTask

kafkaClusterIdToUrlMap, LEADER_OFFSET_LAG_FILTER, storeDeserializerCache, VALID_LAG, veniceWriter, viewWriters

Fields inherited from class com.linkedin.davinci.kafka.consumer.StoreIngestionTask

aggKafkaConsumerService, amplificationFactor, amplificationFactorAdapter, availableSchemaIds, bootstrapTimeoutInMs, chunkAssembler, compressionStrategy, compressor, compressorFactory, consumerActionSequenceNumber, consumerActionsQueue, databaseSyncBytesIntervalForDeferredWriteMode, databaseSyncBytesIntervalForTransactionalMode, dataRecoverySourceVersionNumber, defaultReadyToServeChecker, deserializedSchemaIds, diskUsage, divErrorMetricCallback, emitMetrics, emptyPollSleepMs, errorPartitionId, hostLevelIngestionStats, hybridStoreConfig, idleCounter, ingestionTaskName, isChunked, isCurrentVersion, isDataRecovery, isDaVinciClient, isIsolatedIngestion, isRunning, isWriteComputationEnabled, kafkaClusterUrlResolver, kafkaProps, kafkaVersionTopic, KILL_WAIT_TIME_MS, localKafkaClusterId, localKafkaServer, localKafkaServerSingletonSet, metaStoreWriter, partitionConsumptionStateMap, partitionToPendingConsumerActionCountMap, pubSubTopicRepository, readCycleDelayMs, readOnlyForBatchOnlyStoreEnabled, realTimeTopic, recordLevelMetricEnabled, REDUNDANT_LOGGING_FILTER, SCHEMA_POLLING_DELAY_MS, schemaRepository, serverConfig, statusReportAdapter, storageEngine, storageEngineRepository, storageMetadataService, STORE_VERSION_POLLING_DELAY_MS, storeBufferService, storeName, storeRepository, storeVersionPartitionCount, subPartitionCount, topicManagerRepository, venicePartitioner, versionedDIVStats, versionedIngestionStats, versionNumber, versionTopic, writeComputeFailureCode

Constructor Summary

Constructors

Constructor

Description

ActiveActiveStoreIngestionTask(StoreIngestionTaskFactory.Builder builder, Store store, Version version, java.util.Properties kafkaConsumerProperties, java.util.function.BooleanSupplier isCurrentVersion, VeniceStoreVersionConfig storeConfig, int errorPartitionId, boolean isIsolatedIngestion, java.util.Optional<ObjectCacheBackend> cacheBackend, java.util.function.Function<java.lang.Integer,DaVinciRecordTransformer> getRecordTransformer)

Method Summary

Modifier and Type	Method	Description
protected java.util.Map<java.lang.String,java.lang.Long>	calculateLeaderUpstreamOffsetWithTopicSwitch(PartitionConsumptionState partitionConsumptionState, TopicSwitch topicSwitch, PubSubTopic newSourceTopic, java.util.List<java.lang.CharSequence> unreachableBrokerList)
protected LeaderProducerCallback	createProducerCallback(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord, PartitionConsumptionState partitionConsumptionState, LeaderProducedRecordContext leaderProducedRecordContext, int subPartition, java.lang.String kafkaUrl, long beforeProcessingRecordTimestampNs)
protected StoreIngestionTask.DelegateConsumerRecordResult	delegateConsumerRecord(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord, int subPartition, java.lang.String kafkaUrl, int kafkaClusterId, long beforeProcessingPerRecordTimestampNs, long beforeProcessingBatchRecordsTimestampMs)	The goal of this function is to possibly produce the incoming kafka message consumed from local VT, remote VT, RT or SR topic to local VT if needed.
protected long	getLatestConsumedUpstreamOffsetForHybridOffsetLagMeasurement(PartitionConsumptionState pcs, java.lang.String upstreamKafkaUrl)	Different from the persisted upstream offset map in OffsetRecord, latest consumed upstream offset map is maintained for each individual Kafka url.
protected long	getLatestPersistedUpstreamOffsetForHybridOffsetLagMeasurement(PartitionConsumptionState pcs, java.lang.String upstreamKafkaUrl)	For A/A, there are multiple entries in upstreamOffsetMap during RT ingestion.
protected java.util.function.BiConsumer<ChunkAwareCallback,LeaderMetadataWrapper>	getProduceToTopicFunction(byte[] key, java.nio.ByteBuffer updatedValueBytes, java.nio.ByteBuffer updatedRmdBytes, ChunkedValueManifest oldValueManifest, ChunkedValueManifest oldRmdManifest, int valueSchemaId, boolean resultReuseInput)
long	getRegionHybridOffsetLag(int regionId)
RmdSerDe	getRmdSerDe()
boolean	isReadyToServeAnnouncedWithRTLag()	used for metric purposes
protected boolean	isRealTimeBufferReplayStarted(PartitionConsumptionState partitionConsumptionState)
boolean	isTransientRecordBufferUsed()	For Active-Active this buffer is always used.
protected void	leaderExecuteTopicSwitch(PartitionConsumptionState partitionConsumptionState, TopicSwitch topicSwitch, PubSubTopic newSourceTopic)
protected long	measureRTOffsetLagForMultiRegions(java.util.Set<java.lang.String> sourceRealTimeTopicKafkaURLs, PartitionConsumptionState partitionConsumptionState, boolean shouldLogLag)	For stores in aggregate mode this is optimistic and returns the minimum lag of all fabric.
protected void	processMessageAndMaybeProduceToKafka(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord, PartitionConsumptionState partitionConsumptionState, int subPartition, java.lang.String kafkaUrl, int kafkaClusterId, long beforeProcessingRecordTimestampNs, long currentTimeForMetricsMs)
protected boolean	processTopicSwitch(ControlMessage controlMessage, int partition, long offset, PartitionConsumptionState partitionConsumptionState)	Process TopicSwitch control message at given partition offset for a specific PartitionConsumptionState.
protected void	produceToLocalKafka(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord, PartitionConsumptionState partitionConsumptionState, LeaderProducedRecordContext leaderProducedRecordContext, java.util.function.BiConsumer<ChunkAwareCallback,LeaderMetadataWrapper> produceFunction, int subPartition, java.lang.String kafkaUrl, int kafkaClusterId, long beforeProcessingRecordTimestampNs)
protected void	putInStorageEngine(int partition, byte[] keyBytes, Put put)	Persist Put record to storage engine.
protected void	removeFromStorageEngine(int partition, byte[] keyBytes, Delete delete)
protected boolean	shouldCheckLeaderCompleteStateInFollower()	For non AA hybrid stores with AGGREGATE DRP, SIT reads from parent RT while the HB is written to the child RTs.
protected void	startConsumingAsLeader(PartitionConsumptionState partitionConsumptionState)
protected void	updateLatestInMemoryLeaderConsumedRTOffset(PartitionConsumptionState pcs, java.lang.String kafkaUrl, long offset)
protected void	updateLatestInMemoryProcessedOffset(PartitionConsumptionState partitionConsumptionState, PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord, LeaderProducedRecordContext leaderProducedRecordContext, java.lang.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.

Methods inherited from class com.linkedin.davinci.kafka.consumer.LeaderFollowerStoreIngestionTask

addPartitionConsumptionState, checkAndHandleUpstreamOffsetRewind, checkAndLogIfLagIsAcceptableForHybridStore, checkLongRunningTaskState, closeVeniceViewWriters, closeVeniceWriters, consumerUnSubscribeAllTopics, demoteToStandby, getAndUpdateLeaderCompletedState, getBatchFollowerOffsetLag, getBatchLeaderOffsetLag, getBatchReplicationLag, getConsumptionSourceKafkaAddress, getFollowerOffsetLag, getHybridFollowerOffsetLag, getHybridLeaderOffsetLag, getLeaderOffsetLag, getRealTimeDataSourceKafkaAddress, getSubPartitionId, getTopicPartitionOffsetByKafkaURL, getVeniceWriter, getWriteComputeErrorCode, isHybridFollower, isLeader, maybeCompressData, maybeSendIngestionHeartbeat, measureHybridOffsetLag, measureRTOffsetLagForSingleRegion, processConsumerAction, processVersionSwapMessage, promoteToLeader, recordHeartbeatReceived, recordProcessedRecordStats, recordWriterStats, reportIfCatchUpVersionTopicOffset, shouldCompressData, shouldNewLeaderSwitchToRemoteConsumption, shouldPersistRecord, shouldProcessRecord, shouldProduceToVersionTopic, startConsumingAsLeaderInTransitionFromStandby, syncConsumedUpstreamRTOffsetMapIfNeeded, syncTopicSwitchToIngestionMetadataService, updateLeaderTopicOnFollower, updateOffsetMetadataInOffsetRecord, updateOffsetsFromConsumerRecord, waitForAllMessageToBeProcessedFromTopicPartition, waitForLastLeaderPersistFuture

Methods inherited from class com.linkedin.davinci.kafka.consumer.StoreIngestionTask

checkIngestionProgress, cloneProducerStates, close, consumerBatchUnsubscribe, consumerHasAnySubscription, consumerHasSubscription, consumerResetOffset, consumerSubscribe, consumerUnSubscribe, createKafkaConsumerProperties, disableMetricsEmission, dumpPartitionConsumptionStates, dumpStoreVersionState, enableMetricsEmission, getAmplificationFactor, getCompressionStrategy, getCompressor, getFailedIngestionPartitionCount, getHostLevelIngestionStats, getIngestionTaskName, getKafkaVersionTopic, getLatestConsumedProducerTimestampWithSubPartition, getLeaderState, getOffsetToOnlineLagThresholdPerPartition, getPartitionConsumptionState, getPartitionOffsetLagBasedOnMetrics, getSchemaRepo, getServerConfig, getStatusReportAdapter, getStorageEngine, getStoreName, getTopicManager, getTopicPartitionEndOffSet, getVersionedDIVStats, getVersionIngestionStats, getVersionNumber, getVersionTopic, hasAnyPartitionConsumptionState, hasAnySubscription, hasPendingPartitionIngestionAction, isActiveActiveReplicationEnabled, isChunked, isFutureVersion, isHybridMode, isIngestionTaskActive, isMetricsEmissionEnabled, isPartitionConsumingOrHasPendingIngestionAction, isProducingVersionTopicHealthy, isReadyToServe, isRunning, isSegmentControlMsg, isStuckByMemoryConstraint, isUserSystemStore, kill, logStorageOperationWhileUnsubscribed, measureLagWithCallToPubSub, measureLagWithCallToPubSub, minZeroLag, nextSeqNum, processCommonConsumerAction, processConsumerRecord, processEndOfIncrementalPush, processEndOfPush, processStartOfIncrementalPush, produceToStoreBufferService, produceToStoreBufferServiceOrKafka, recordChecksumVerificationFailure, reportError, resetPartitionConsumptionOffset, resolveSourceKafkaServersWithinTopicSwitch, run, setLastConsumerException, setLastStoreIngestionException, setPartitionConsumptionState, shouldUpdateUpstreamOffset, shutdown, subscribePartition, subscribePartition, throwIfNotRunning, throwOrLogStorageFailureDependingIfStillSubscribed, unSubscribePartition, unSubscribePartition, updateOffsetMetadataAndSync, validateMessage, waitVersionStateAvailable

Methods inherited from class java.lang.Object

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

Constructor Detail

ActiveActiveStoreIngestionTask

public ActiveActiveStoreIngestionTask(StoreIngestionTaskFactory.Builder builder,
                                      Store store,
                                      Version version,
                                      java.util.Properties kafkaConsumerProperties,
                                      java.util.function.BooleanSupplier isCurrentVersion,
                                      VeniceStoreVersionConfig storeConfig,
                                      int errorPartitionId,
                                      boolean isIsolatedIngestion,
                                      java.util.Optional<ObjectCacheBackend> cacheBackend,
                                      java.util.function.Function<java.lang.Integer,DaVinciRecordTransformer> getRecordTransformer)

Method Detail

delegateConsumerRecord

protected StoreIngestionTask.DelegateConsumerRecordResult delegateConsumerRecord(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord,
                                                                                 int subPartition,
                                                                                 java.lang.String kafkaUrl,
                                                                                 int kafkaClusterId,
                                                                                 long beforeProcessingPerRecordTimestampNs,
                                                                                 long beforeProcessingBatchRecordsTimestampMs)

Description copied from class: LeaderFollowerStoreIngestionTask

The goal of this function is to possibly produce the incoming kafka message consumed from local VT, remote VT, RT or SR topic to local VT if needed. It's decided based on the function output of LeaderFollowerStoreIngestionTask.shouldProduceToVersionTopic(com.linkedin.davinci.kafka.consumer.PartitionConsumptionState) and message type. It also perform any necessary additional computation operation such as for write-compute/update message. It returns a boolean indicating if it was produced to kafka or not. This function should be called as one of the first steps in processing pipeline for all messages consumed from any kafka topic. The caller of this function should only process this {@param consumerRecord} further if the return is StoreIngestionTask.DelegateConsumerRecordResult.QUEUED_TO_DRAINER. This function assumes LeaderFollowerStoreIngestionTask.shouldProcessRecord(PubSubMessage, int) has been called which happens in StoreIngestionTask.produceToStoreBufferServiceOrKafka(Iterable, PubSubTopicPartition, String, int) before calling this and the it was decided that this record needs to be processed. It does not perform any validation check on the PartitionConsumptionState object to keep the goal of the function simple and not overload. Also DIV validation is done here if the message is received from RT topic. For more info please see please see StoreIngestionTask#internalProcessConsumerRecord(PubSubMessage, PartitionConsumptionState, LeaderProducedRecordContext, int, String, long) This function may modify the original record in KME and it is unsafe to use the payload from KME directly after this function.

Overrides:

delegateConsumerRecord in class LeaderFollowerStoreIngestionTask

Returns:

a StoreIngestionTask.DelegateConsumerRecordResult indicating what to do with the record

putInStorageEngine

protected void putInStorageEngine(int partition,
                                  byte[] keyBytes,
                                  Put put)

Description copied from class: StoreIngestionTask

Persist Put record to storage engine.

Overrides:

putInStorageEngine in class StoreIngestionTask

removeFromStorageEngine

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

Overrides:

removeFromStorageEngine in class StoreIngestionTask

getRmdSerDe

public RmdSerDe getRmdSerDe()

processMessageAndMaybeProduceToKafka

protected void processMessageAndMaybeProduceToKafka(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord,
                                                    PartitionConsumptionState partitionConsumptionState,
                                                    int subPartition,
                                                    java.lang.String kafkaUrl,
                                                    int kafkaClusterId,
                                                    long beforeProcessingRecordTimestampNs,
                                                    long currentTimeForMetricsMs)

Overrides:

processMessageAndMaybeProduceToKafka in class LeaderFollowerStoreIngestionTask

produceToLocalKafka

protected void produceToLocalKafka(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord,
                                   PartitionConsumptionState partitionConsumptionState,
                                   LeaderProducedRecordContext leaderProducedRecordContext,
                                   java.util.function.BiConsumer<ChunkAwareCallback,LeaderMetadataWrapper> produceFunction,
                                   int subPartition,
                                   java.lang.String kafkaUrl,
                                   int kafkaClusterId,
                                   long beforeProcessingRecordTimestampNs)

Overrides:

produceToLocalKafka in class LeaderFollowerStoreIngestionTask

calculateLeaderUpstreamOffsetWithTopicSwitch

protected java.util.Map<java.lang.String,java.lang.Long> calculateLeaderUpstreamOffsetWithTopicSwitch(PartitionConsumptionState partitionConsumptionState,
                                                                                                            TopicSwitch topicSwitch,
                                                                                                            PubSubTopic newSourceTopic,
                                                                                                            java.util.List<java.lang.CharSequence> unreachableBrokerList)

Overrides:

calculateLeaderUpstreamOffsetWithTopicSwitch in class LeaderFollowerStoreIngestionTask

startConsumingAsLeader

protected void startConsumingAsLeader(PartitionConsumptionState partitionConsumptionState)

Overrides:

startConsumingAsLeader in class LeaderFollowerStoreIngestionTask

leaderExecuteTopicSwitch

protected void leaderExecuteTopicSwitch(PartitionConsumptionState partitionConsumptionState,
                                        TopicSwitch topicSwitch,
                                        PubSubTopic newSourceTopic)

Overrides:

leaderExecuteTopicSwitch in class LeaderFollowerStoreIngestionTask

processTopicSwitch

protected boolean processTopicSwitch(ControlMessage controlMessage,
                                     int partition,
                                     long offset,
                                     PartitionConsumptionState partitionConsumptionState)

Process TopicSwitch control message at given partition offset for a specific PartitionConsumptionState. Return whether we need to execute additional ready-to-serve check after this message is processed.

Overrides:

processTopicSwitch in class LeaderFollowerStoreIngestionTask

updateLatestInMemoryProcessedOffset

protected void updateLatestInMemoryProcessedOffset(PartitionConsumptionState partitionConsumptionState,
                                                   PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord,
                                                   LeaderProducedRecordContext leaderProducedRecordContext,
                                                   java.lang.String kafkaUrl,
                                                   boolean dryRun)

Description copied from class: StoreIngestionTask

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.

Overrides:

updateLatestInMemoryProcessedOffset in class LeaderFollowerStoreIngestionTask

isRealTimeBufferReplayStarted

protected boolean isRealTimeBufferReplayStarted(PartitionConsumptionState partitionConsumptionState)

Overrides:

isRealTimeBufferReplayStarted in class LeaderFollowerStoreIngestionTask

getLatestPersistedUpstreamOffsetForHybridOffsetLagMeasurement

protected long getLatestPersistedUpstreamOffsetForHybridOffsetLagMeasurement(PartitionConsumptionState pcs,
                                                                             java.lang.String upstreamKafkaUrl)

For A/A, there are multiple entries in upstreamOffsetMap during RT ingestion. If the current DataReplicationPolicy is on Aggregate mode, A/A will check the upstream offset lags from all regions; otherwise, only check the upstream offset lag from the local region.

Overrides:

getLatestPersistedUpstreamOffsetForHybridOffsetLagMeasurement in class LeaderFollowerStoreIngestionTask

getLatestConsumedUpstreamOffsetForHybridOffsetLagMeasurement

protected long getLatestConsumedUpstreamOffsetForHybridOffsetLagMeasurement(PartitionConsumptionState pcs,
                                                                            java.lang.String upstreamKafkaUrl)

Different from the persisted upstream offset map in OffsetRecord, latest consumed upstream offset map is maintained for each individual Kafka url.

Overrides:

getLatestConsumedUpstreamOffsetForHybridOffsetLagMeasurement in class LeaderFollowerStoreIngestionTask

updateLatestInMemoryLeaderConsumedRTOffset

protected void updateLatestInMemoryLeaderConsumedRTOffset(PartitionConsumptionState pcs,
                                                          java.lang.String kafkaUrl,
                                                          long offset)

Overrides:

updateLatestInMemoryLeaderConsumedRTOffset in class LeaderFollowerStoreIngestionTask

isTransientRecordBufferUsed

public boolean isTransientRecordBufferUsed()

For Active-Active this buffer is always used.

Overrides:

isTransientRecordBufferUsed in class StoreIngestionTask

Returns:

shouldCheckLeaderCompleteStateInFollower

protected boolean shouldCheckLeaderCompleteStateInFollower()

Description copied from class: LeaderFollowerStoreIngestionTask

For non AA hybrid stores with AGGREGATE DRP, SIT reads from parent RT while the HB is written to the child RTs. Once all hybrid stores are either AA for cross colo replication and non AA otherwise, DRP and this extra check can also be removed.

Overrides:

shouldCheckLeaderCompleteStateInFollower in class LeaderFollowerStoreIngestionTask

getRegionHybridOffsetLag

public long getRegionHybridOffsetLag(int regionId)

Overrides:

getRegionHybridOffsetLag in class LeaderFollowerStoreIngestionTask

measureRTOffsetLagForMultiRegions

protected long measureRTOffsetLagForMultiRegions(java.util.Set<java.lang.String> sourceRealTimeTopicKafkaURLs,
                                                 PartitionConsumptionState partitionConsumptionState,
                                                 boolean shouldLogLag)

For stores in aggregate mode this is optimistic and returns the minimum lag of all fabric. This is because in aggregate mode duplicate msg consumption happen from all fabric. So it should be fine to consider the lowest lag. For stores in active/active mode, if no fabric is unreachable, return the maximum lag of all fabrics. If only one fabric is unreachable, return the maximum lag of other fabrics. If more than one fabrics are unreachable, return Long.MAX_VALUE, which means the partition is not ready-to-serve. TODO: For active/active incremental push stores or stores with only one samza job, we should consider the weight of unreachable fabric and make the decision. For example, we should not let partition ready-to-serve when the only source fabric is unreachable. In non-aggregate mode of consumption only return the local fabric lag

Overrides:

measureRTOffsetLagForMultiRegions in class LeaderFollowerStoreIngestionTask

Parameters:

sourceRealTimeTopicKafkaURLs -

partitionConsumptionState -

shouldLogLag -

Returns:

isReadyToServeAnnouncedWithRTLag

public boolean isReadyToServeAnnouncedWithRTLag()

used for metric purposes

Overrides:

isReadyToServeAnnouncedWithRTLag in class LeaderFollowerStoreIngestionTask

getProduceToTopicFunction

protected java.util.function.BiConsumer<ChunkAwareCallback,LeaderMetadataWrapper> getProduceToTopicFunction(byte[] key,
                                                                                                                  java.nio.ByteBuffer updatedValueBytes,
                                                                                                                  java.nio.ByteBuffer updatedRmdBytes,
                                                                                                                  ChunkedValueManifest oldValueManifest,
                                                                                                                  ChunkedValueManifest oldRmdManifest,
                                                                                                                  int valueSchemaId,
                                                                                                                  boolean resultReuseInput)

createProducerCallback

protected LeaderProducerCallback createProducerCallback(PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long> consumerRecord,
                                                        PartitionConsumptionState partitionConsumptionState,
                                                        LeaderProducedRecordContext leaderProducedRecordContext,
                                                        int subPartition,
                                                        java.lang.String kafkaUrl,
                                                        long beforeProcessingRecordTimestampNs)

Overrides:

createProducerCallback in class LeaderFollowerStoreIngestionTask