Package com.linkedin.davinci.validation

Class DataIntegrityValidator

java.lang.Object
com.linkedin.davinci.validation.DataIntegrityValidator
public class DataIntegrityValidator extends Object
Colloquially known as DIV (Data Integrity Validator). This class is a library that can validate the topic message during consumption, which can be used in Venice-Server/Da-Vinci and ETL. In high level, it keeps track of messages produced by different producers, and validates data integrity in 4 perspectives: 1. Whether a segment starts from a non-zero sequence number (UNREGISTERED_PRODUCER); 2. Whether there is a gap between segments or within a segment (MISSING); 3. Whether data within a segment is corrupted (CORRUPT); 4. Whether producers have produced duplicate messages, which is fine and expected due to producer retries (DUPLICATE).

  • Field Details

  • Constructor Details

    • DataIntegrityValidator

      public DataIntegrityValidator(String topicName, PubSubPositionDeserializer pubSubPositionDeserializer)

    • DataIntegrityValidator

      public DataIntegrityValidator(String topicName, PubSubPositionDeserializer pubSubPositionDeserializer, long logCompactionDelayInMs)
      This constructor is used by a proprietary ETL project. Do not clean up (yet)! TODO: Open source the ETL or make it stop depending on an exotic open source API

    • DataIntegrityValidator

      public DataIntegrityValidator(String topicName, PubSubPositionDeserializer pubSubPositionDeserializer, long logCompactionDelayInMs, long maxAgeInMs)

  • Method Details

    • clearPartition

      public void clearPartition(int partition)
      In some cases, such as when resetting offsets or unsubscribing from a partition, the PartitionTracker should forget about the state that it accumulated for a given partition.
      Parameters:
      partition - to clear state for

    • clearRtSegments

      public void clearRtSegments(int partition)

    • setPartitionState

      public void setPartitionState(PartitionTracker.TopicType type, int partition, OffsetRecord offsetRecord)

    • setPartitionState

      public void setPartitionState(PartitionTracker.TopicType type, int partition, Map<CharSequence,ProducerPartitionState> producerPartitionStateMap)

    • validateMessage

      public void validateMessage(PartitionTracker.TopicType type, DefaultPubSubMessage consumerRecord, boolean endOfPushReceived, Lazy<Boolean> tolerateMissingMsgs) throws DataValidationException
      Run a thorough DIV check on the message, including UNREGISTERED_PRODUCER, MISSING, CORRUPT and DUPLICATE.
      Throws:
      DataValidationException

    • updateOffsetRecordForPartition

      public void updateOffsetRecordForPartition(PartitionTracker.TopicType type, int partition, OffsetRecord offsetRecord)
      For a given partition, find all the producers that has written to this partition and update the offsetRecord using segment information. Prior to this, the state which is expired according to maxAgeInMs will be cleared.
      Parameters:
      partition - to extract info for
      offsetRecord - to modify

    • cloneVtProducerStates

      public void cloneVtProducerStates(int partition, DataIntegrityValidator newValidator)

    • cloneRtProducerStates

      public PartitionTracker cloneRtProducerStates(int partition, String brokerUrl)
      Returns the RT DIV state for a given partition and broker URL pair.

    • cloneVtProducerStates

      public PartitionTracker cloneVtProducerStates(int partition)
      Returns the VT DIV state and latest consumed vt offset for a given partition.

    • checkMissingMessage

      public void checkMissingMessage(DefaultPubSubMessage consumerRecord, Optional<PartitionTracker.DIVErrorMetricCallback> errorMetricCallback) throws DataValidationException
      Only check for missing sequence number; segment starting from a positive sequence number is acceptable considering real-time buffer replay would start in the middle of a segment; checksum is also ignored for the same reason. If missing message happens for an old message that is older than the log compaction lag threshold, MISSING exception will not be thrown because it's expected that log compaction would compact old messages. However, if data are fresh and missing message is detected, MISSING exception will be thrown. This API is used by a proprietary ETL project. Do not clean up (yet)! TODO: Open source the ETL or make it stop depending on an exotic open source API
      Throws:
      DataValidationException

    • updateLatestConsumedVtPosition

      public void updateLatestConsumedVtPosition(int partition, PubSubPosition vtPosition)

    • hasGlobalRtDivState

      public boolean hasGlobalRtDivState(int partition)

    • hasVtDivState

      public boolean hasVtDivState(int partition)