[Controller][all] Do not truncate Kafka topic immediately for fatal d…

…ata validation error before EOP (#937)

When the ingestion of a new store version is failed, today, we truncate the Kafka topic of the store version by updating its retention time to a small value (15 seconds), specified by DEPRECATED_TOPIC_RETENTION_MS.

This is fine for regular failures but for fatal data validation errors, which indicates critical issues happened during the batch push period (before EOP), truncating the topic too early can prevent us from finding the root cause, as the Kafka data is gone.

This change adjusts the Kafka topic retention time (to 2 days) when fatal data validation errors is identified, so that we can have enough time to investigate.

Meanwhile, there is additional logic added to the TopicCleanupService, i.e. even if a topic's retention time is > DEPRECATED_TOPIC_MAX_RETENTION_MS, it can still be consider for deletion:

- If topic retention is 2 (DEPRECATED_TOPIC_RETENTION_MS) days.
- If The topic is a version topic.
- Get topic creation time (from venice_system_store_push_job_details_store) and check it's already more than 2 days (DEPRECATED_TOPIC_RETENTION_MS), if yes, delete it.

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -9,6 +9,7 @@
 import static com.linkedin.venice.ConfigKeys.KAFKA_BOOTSTRAP_SERVERS;
 import static com.linkedin.venice.LogMessages.KILLED_JOB_MESSAGE;
 import static com.linkedin.venice.kafka.protocol.enums.ControlMessageType.START_OF_SEGMENT;
+import static com.linkedin.venice.utils.Utils.FATAL_DATA_VALIDATION_ERROR;
 import static java.util.concurrent.TimeUnit.HOURS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
@@ -2182,11 +2183,11 @@ public void processConsumerRecord(
       int faultyPartition = record.getTopicPartition().getPartitionNumber();
       String errorMessage;
       if (amplificationFactor != 1 && record.getTopicPartition().getPubSubTopic().isRealTime()) {
-        errorMessage = "Fatal data validation problem with in RT topic partition " + faultyPartition + ", offset "
+        errorMessage = FATAL_DATA_VALIDATION_ERROR + " with in RT topic partition " + faultyPartition + ", offset "
             + record.getOffset() + ", leaderSubPartition: " + subPartition;
       } else {
         errorMessage =
-            "Fatal data validation problem with partition " + faultyPartition + ", offset " + record.getOffset();
+            FATAL_DATA_VALIDATION_ERROR + " with partition " + faultyPartition + ", offset " + record.getOffset();
       }
       // TODO need a way to safeguard DIV errors from backup version that have once been current (but not anymore)
       // during re-balancing

internal/venice-client-common/src/main/java/com/linkedin/venice/pushmonitor/ExecutionStatus.java

@@ -217,4 +217,8 @@ public static boolean isError(String errorString) {
       return false;
     }
   }
+
+  public static boolean isError(ExecutionStatus status) {
+    return status == ERROR;
+  }
 }

internal/venice-client-common/src/test/java/com/linkedin/venice/pushmonitor/ExecutionStatusTest.java

@@ -31,6 +31,7 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
 
@@ -183,4 +184,15 @@ public void testRootStatus() {
       assertEquals(status.getRootStatus(), rootStatusMap.getOrDefault(status, status));
     }
   }
+
+  @Test
+  public void testErrorExecutionStatus() {
+    for (ExecutionStatus status: ExecutionStatus.values()) {
+      if (status == ERROR) {
+        Assert.assertTrue(ExecutionStatus.isError(status));
+      } else {
+        Assert.assertFalse(ExecutionStatus.isError(status));
+      }
+    }
+  }
 }

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -242,6 +242,9 @@ private ConfigKeys() {
    */
   public static final String DEPRECATED_TOPIC_RETENTION_MS = "deprecated.topic.retention.ms";
 
+  public static final String FATAL_DATA_VALIDATION_FAILURE_TOPIC_RETENTION_MS =
+      "fatal.data.validation.failure.topic.retention.ms";
+
   /**
    * This config is to indicate the max retention policy we have setup for deprecated jobs currently and in the past.
    * And this is used to decide whether the topic is deprecated or not during topic cleanup.

internal/venice-common/src/main/java/com/linkedin/venice/pushmonitor/OfflinePushStatus.java

@@ -281,6 +281,19 @@ public String getLatestIncrementalPushVersion(PartitionAssignment partitionAssig
     return latestIncrementalPushVersion;
   }
 
+  public boolean hasFatalDataValidationError() {
+    return partitionIdToStatus.values().stream().anyMatch(partitionStatus -> {
+      if (partitionStatus.hasFatalDataValidationError()) {
+        LOGGER.warn(
+            "Fatal data validation error found in topic: {}, partition: {}",
+            kafkaTopic,
+            partitionStatus.getPartitionId());
+        return true;
+      }
+      return false;
+    });
+  }
+
   public String getKafkaTopic() {
     return kafkaTopic;
   }

internal/venice-common/src/main/java/com/linkedin/venice/pushmonitor/PartitionStatus.java

@@ -1,6 +1,7 @@
 package com.linkedin.venice.pushmonitor;
 
 import static com.linkedin.venice.pushmonitor.ExecutionStatus.NOT_CREATED;
+import static com.linkedin.venice.utils.Utils.*;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -86,6 +87,16 @@ public List<StatusSnapshot> getReplicaHistoricStatusList(String instanceId) {
     return replicaStatus.getStatusHistory();
   }
 
+  public boolean hasFatalDataValidationError() {
+    for (ReplicaStatus replicaStatus: replicaStatusMap.values()) {
+      if (ExecutionStatus.isError(replicaStatus.getCurrentStatus())
+          && replicaStatus.getIncrementalPushVersion().contains(FATAL_DATA_VALIDATION_ERROR)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

internal/venice-common/src/main/java/com/linkedin/venice/utils/Utils.java

@@ -76,6 +76,8 @@ public class Utils {
   public static final String NEW_LINE_CHAR = System.lineSeparator();
   public static final AtomicBoolean SUPPRESS_SYSTEM_EXIT = new AtomicBoolean();
 
+  public static final String FATAL_DATA_VALIDATION_ERROR = "fatal data validation problem";
+
   /**
    * Print an error and exit with error code 1
    *

internal/venice-common/src/test/java/com/linkedin/venice/pushmonitor/OfflinePushStatusTest.java

@@ -22,6 +22,7 @@
 import static com.linkedin.venice.pushmonitor.ExecutionStatus.TOPIC_SWITCH_RECEIVED;
 import static com.linkedin.venice.pushmonitor.ExecutionStatus.UNKNOWN;
 import static com.linkedin.venice.pushmonitor.ExecutionStatus.WARNING;
+import static com.linkedin.venice.utils.Utils.FATAL_DATA_VALIDATION_ERROR;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
@@ -30,6 +31,7 @@
 import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.meta.OfflinePushStrategy;
 import com.linkedin.venice.meta.PartitionAssignment;
+import com.linkedin.venice.utils.DataProviderUtils;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
@@ -272,6 +274,46 @@ public void testGetStatusUpdateTimestamp() {
     assertEquals(statusUpdateTimestamp, Long.valueOf(startOfProgress.toEpochSecond(ZoneOffset.UTC)));
   }
 
+  @Test(dataProviderClass = DataProviderUtils.class, dataProvider = "True-and-False")
+  public void testHasFatalDataValidationError(boolean hasFatalDataValidationError) {
+    String kafkaTopic = "testTopic";
+    int numberOfPartition = 3;
+    int replicationFactor = 2;
+    OfflinePushStrategy strategy = OfflinePushStrategy.WAIT_N_MINUS_ONE_REPLCIA_PER_PARTITION;
+
+    // Create an OfflinePushStatus object
+    OfflinePushStatus offlinePushStatus =
+        new OfflinePushStatus(kafkaTopic, numberOfPartition, replicationFactor, strategy);
+
+    // Create a PartitionStatus
+    PartitionStatus partitionStatus = new PartitionStatus(1);
+
+    // Create a ReplicaStatus with an error ExecutionStatus and an incrementalPushVersion that contains "Fatal data
+    // validation problem"
+    ReplicaStatus replicaStatus = new ReplicaStatus("instance1", true);
+    if (hasFatalDataValidationError) {
+      replicaStatus.updateStatus(ExecutionStatus.ERROR);
+      replicaStatus.setIncrementalPushVersion(
+          FATAL_DATA_VALIDATION_ERROR + " with partition 1, offset 1096534. Consumption will be halted.");
+    } else {
+      replicaStatus.updateStatus(ExecutionStatus.COMPLETED);
+    }
+
+    // Add the ReplicaStatus to the replicaStatusMap of the PartitionStatus
+    partitionStatus.setReplicaStatuses(Collections.singleton(replicaStatus));
+
+    // Set the PartitionStatus in the OfflinePushStatus
+    offlinePushStatus.setPartitionStatus(partitionStatus);
+
+    if (hasFatalDataValidationError) {
+      // Assert that hasFatalDataValidationError returns true
+      Assert.assertTrue(offlinePushStatus.hasFatalDataValidationError());
+    } else {
+      // Assert that hasFatalDataValidationError returns false
+      Assert.assertFalse(offlinePushStatus.hasFatalDataValidationError());
+    }
+  }
+
   private void testValidTargetStatus(ExecutionStatus from, ExecutionStatus to) {
     OfflinePushStatus offlinePushStatus =
         new OfflinePushStatus(kafkaTopic, numberOfPartition, replicationFactor, strategy);