[controller][vpj][protocol] Propagate KILL job trigger and details from VPJ

to servers

Add support for propagating the real KILL caller information (trigger reason
+ details)
from VenicePushJob through Controllers to Servers, enabling better
observability and
debugging of push job terminations.

Key changes:
- Add PushJobKillTrigger enum with trigger types: USER_REQUEST,
VERSION_RETIREMENT,
  SLA_VIOLATION, PREEMPTED_BY_FULL_PUSH, INGESTION_FAILURE,
VERSION_CREATION_FAILURE,
  PUSH_JOB_FAILED, LINGERING_VERSION_TOPIC, UNKNOWN
- Add AdminOperation v96 schema with trigger/details fields in
KillOfflinePushJob
- Add ParticipantMessageValue v2 schema with trigger/details fields in
KillPushJob
- Update Admin.killOfflinePush() and StoreCleaner.deleteOneStoreVersion()
interfaces
- Propagate trigger/details through VeniceHelixAdmin and
VeniceParentHelixAdmin
- Add PUSH_JOB_KILL_TRIGGER and PUSH_JOB_KILL_DETAILS controller API params
- VPJ uses SLA_VIOLATION for timeout kills, PUSH_JOB_FAILED for error cleanup
- Add SLA_TIMEOUT error type

Author: sushantmane

build.gradle

@@ -328,7 +328,7 @@ subprojects {
       // project(':internal:venice-common').file('src/main/resources/avro/KafkaMessageEnvelope/v12', PathValidation.DIRECTORY)
       def versionOverrides = [
           project(':internal:venice-common').file('src/main/resources/avro/StoreMetaValue/v39', PathValidation.DIRECTORY),
-          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v94', PathValidation.DIRECTORY)
+          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v96', PathValidation.DIRECTORY)
       ]
 
       def schemaDirs = [sourceDir]

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/VenicePushJob.java

@@ -133,6 +133,7 @@
 import com.linkedin.venice.meta.Version;
 import com.linkedin.venice.meta.VersionStatus;
 import com.linkedin.venice.meta.ViewConfig;
+import com.linkedin.venice.participant.protocol.enums.PushJobKillTrigger;
 import com.linkedin.venice.partitioner.DefaultVenicePartitioner;
 import com.linkedin.venice.partitioner.VenicePartitioner;
 import com.linkedin.venice.pushmonitor.ExecutionStatus;
@@ -987,11 +988,14 @@ private void setupJobTimeoutMonitor() {
     }
 
     LOGGER.info("Scheduling timeout executor for store: {} with timeout: {}ms", pushJobSetting.storeName, timeoutMs);
+    final long capturedTimeoutMs = timeoutMs;
     timeoutExecutor.schedule(() -> {
-      cancel();
+      String timeoutDetails = "Push job exceeded timeout: " + capturedTimeoutMs + " ms ("
+          + TimeUnit.MILLISECONDS.toHours(capturedTimeoutMs) + " hours)";
+      cancel(PushJobKillTrigger.SLA_VIOLATION, timeoutDetails);
       throw new VeniceTimeoutException(
-          "Failing push-job for store " + pushJobSetting.storeName + " which is still running after " + timeoutMs
-              + " ms (" + TimeUnit.MILLISECONDS.toHours(timeoutMs) + " hours)");
+          "Failing push-job for store " + pushJobSetting.storeName + " which is still running after "
+              + capturedTimeoutMs + " ms (" + TimeUnit.MILLISECONDS.toHours(capturedTimeoutMs) + " hours)");
     }, timeoutMs, TimeUnit.MILLISECONDS);
   }
 
@@ -2916,10 +2920,20 @@ private String pushJobPropertiesToString(
 
   /**
    * A cancel method for graceful cancellation of the running Job to be invoked as a result of user actions or due to
-   * the job exceeding bootstrapToOnlineTimeoutInHours.
+   * the job exceeding bootstrapToOnlineTimeoutInHours. This is the default cancel that doesn't specify a trigger,
+   * which uses USER_REQUEST as the default trigger.
    */
   public void cancel() {
-    killJob(pushJobSetting, controllerClient);
+    cancel(PushJobKillTrigger.USER_REQUEST, null);
+  }
+
+  /**
+   * A cancel method for graceful cancellation of the running Job with a specific trigger reason.
+   * @param trigger the reason for cancellation
+   * @param details additional details about the cancellation
+   */
+  public void cancel(PushJobKillTrigger trigger, String details) {
+    killJob(pushJobSetting, controllerClient, trigger, details);
     if (StringUtils.isEmpty(pushJobSetting.topic)) {
       pushJobDetails.overallStatus.add(getPushJobDetailsStatusTuple(PushJobDetailsStatus.ERROR.getValue()));
     } else {
@@ -2930,6 +2944,14 @@ public void cancel() {
   }
 
   void killJob(PushJobSetting pushJobSetting, ControllerClient controllerClient) {
+    killJob(pushJobSetting, controllerClient, PushJobKillTrigger.PUSH_JOB_FAILED, null);
+  }
+
+  void killJob(
+      PushJobSetting pushJobSetting,
+      ControllerClient controllerClient,
+      PushJobKillTrigger trigger,
+      String details) {
     // Attempting to kill job. There's a race condition, but meh. Better kill when you know it's running
     killDataWriterJob();
     if (!pushJobSetting.isIncrementalPush) {
@@ -2945,11 +2967,16 @@ void killJob(PushJobSetting pushJobSetting, ControllerClient controllerClient) {
       if (StringUtils.isEmpty(pushJobSetting.topic)) {
         LOGGER.error("Could not find a store version to delete for store: {}", pushJobSetting.storeName);
       } else {
+        String triggerStr = trigger != null ? trigger.name() : PushJobKillTrigger.UNKNOWN.name();
         ControllerClient.retryableRequest(
             controllerClient,
             pushJobSetting.controllerRetries,
-            c -> c.killOfflinePushJob(pushJobSetting.topic));
-        LOGGER.info("Offline push job has been killed, topic: {}", pushJobSetting.topic);
+            c -> c.killOfflinePushJob(pushJobSetting.topic, triggerStr, details));
+        LOGGER.info(
+            "Offline push job has been killed, topic: {}, trigger: {}, details: {}",
+            pushJobSetting.topic,
+            trigger,
+            details);
       }
     }
   }

clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/VenicePushJobTest.java

@@ -90,6 +90,7 @@
 import com.linkedin.venice.meta.VersionStatus;
 import com.linkedin.venice.meta.ViewConfig;
 import com.linkedin.venice.meta.ViewConfigImpl;
+import com.linkedin.venice.participant.protocol.enums.PushJobKillTrigger;
 import com.linkedin.venice.partitioner.DefaultVenicePartitioner;
 import com.linkedin.venice.pushmonitor.ExecutionStatus;
 import com.linkedin.venice.schema.AvroSchemaParseUtils;
@@ -601,14 +602,14 @@ public void testDataWriterComputeJobTimeout(Class<? extends DataWriterComputeJob
         doNothing().when(dataWriterJob).validateJob();
         doNothing().when(dataWriterJob).configure(any(), any()); // the spark job takes a long time to configure
         doAnswer(stallDataWriterJob).when(dataWriterJob).runComputeJob();
-        doAnswer(killDataWriterJob).when(pushJob).killJob(any(), any());
+        doAnswer(killDataWriterJob).when(pushJob).killJob(any(), any(), any(PushJobKillTrigger.class), any());
         pushJob.run(); // data writer job will run in this main test thread
       } catch (VeniceException e) {
         // Expected, because the data writer job is not configured to run successfully in this unit test environment
       }
       assertEquals(runningJobLatch.getCount(), 0); // killDataWriterJob() does not occur in the main test thread
       assertEquals(killedJobLatch.getCount(), 0);
-      verify(pushJob, times(1)).cancel();
+      verify(pushJob, times(1)).cancel(eq(PushJobKillTrigger.SLA_VIOLATION), anyString());
       verify(dataWriterJob, times(1)).kill();
       assertEquals(pushJob.getDataWriterComputeJob().getStatus(), DataWriterComputeJob.Status.KILLED);
     }

internal/venice-client-common/src/main/java/com/linkedin/venice/exceptions/ErrorType.java

@@ -16,7 +16,8 @@ public enum ErrorType {
   @JsonEnumDefaultValue
   GENERAL_ERROR(ExceptionType.GENERAL_ERROR), BAD_REQUEST(ExceptionType.BAD_REQUEST),
   CONCURRENT_BATCH_PUSH(ExceptionType.BAD_REQUEST), RESOURCE_STILL_EXISTS(ExceptionType.BAD_REQUEST),
-  PROTOCOL_ERROR(ExceptionType.BAD_REQUEST), ACL_ERROR(ExceptionType.BAD_REQUEST),;
+  PROTOCOL_ERROR(ExceptionType.BAD_REQUEST), ACL_ERROR(ExceptionType.BAD_REQUEST),
+  SLA_TIMEOUT(ExceptionType.GENERAL_ERROR);
 
   private final ExceptionType exceptionType;
 

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerApiConstants.java

@@ -292,4 +292,10 @@ public class ControllerApiConstants {
    * Params for dark cluster
    */
   public static final String STORES_TO_REPLICATE = "stores_to_replicate";
+
+  /**
+   * Params for KILL push job
+   */
+  public static final String PUSH_JOB_KILL_TRIGGER = "push_job_kill_trigger";
+  public static final String PUSH_JOB_KILL_DETAILS = "push_job_kill_details";
 }

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerClient.java

@@ -45,6 +45,8 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.POSITION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_IN_SORTED_ORDER;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_ID;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_KILL_DETAILS;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_KILL_TRIGGER;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_STRATEGY;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_TYPE;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.READ_OPERATION;
@@ -704,12 +706,22 @@ public ControllerResponse rollForwardToFutureVersion(String storeName) {
   }
 
   public ControllerResponse killOfflinePushJob(String kafkaTopic) {
+    return killOfflinePushJob(kafkaTopic, null, null);
+  }
+
+  public ControllerResponse killOfflinePushJob(String kafkaTopic, String trigger, String details) {
     String store = Version.parseStoreFromKafkaTopicName(kafkaTopic);
     int versionNumber = Version.parseVersionFromKafkaTopicName(kafkaTopic);
     QueryParams params = newParams().add(TOPIC, kafkaTopic) // TODO: remove once the controller is deployed to handle
                                                             // store and version instead
         .add(NAME, store)
         .add(VERSION, versionNumber);
+    if (trigger != null) {
+      params.add(PUSH_JOB_KILL_TRIGGER, trigger);
+    }
+    if (details != null) {
+      params.add(PUSH_JOB_KILL_DETAILS, details);
+    }
     return request(ControllerRoute.KILL_OFFLINE_PUSH_JOB, params, ControllerResponse.class);
   }
 

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerRoute.java

@@ -64,6 +64,8 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_DETAILS;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_DURATION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_ID;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_KILL_DETAILS;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_KILL_TRIGGER;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_JOB_STATUS;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_STRATEGY;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.PUSH_TYPE;
@@ -158,8 +160,10 @@ public enum ControllerRoute implements VeniceDimensionInterface {
   UPDATE_DARK_CLUSTER_CONFIG(
       "/update_dark_cluster_config", HttpMethod.POST, Collections.singletonList(CLUSTER), STORES_TO_REPLICATE
   ), JOB("/job", HttpMethod.GET, Arrays.asList(NAME, VERSION)),
-  KILL_OFFLINE_PUSH_JOB("/kill_offline_push_job", HttpMethod.POST, Collections.singletonList(TOPIC)),
-  LIST_STORES("/list_stores", HttpMethod.GET, Collections.emptyList(), INCLUDE_SYSTEM_STORES),
+  KILL_OFFLINE_PUSH_JOB(
+      "/kill_offline_push_job", HttpMethod.POST, Collections.singletonList(TOPIC), PUSH_JOB_KILL_TRIGGER,
+      PUSH_JOB_KILL_DETAILS
+  ), LIST_STORES("/list_stores", HttpMethod.GET, Collections.emptyList(), INCLUDE_SYSTEM_STORES),
   CLEAN_EXECUTION_IDS("/clean_execution_ids", HttpMethod.GET, Collections.emptyList(), CLUSTER),
   LIST_CHILD_CLUSTERS("/list_child_clusters", HttpMethod.GET, Collections.emptyList()),
   LIST_NODES("/list_instances", HttpMethod.GET, Collections.emptyList()),

internal/venice-common/src/main/java/com/linkedin/venice/meta/StoreCleaner.java

@@ -1,7 +1,12 @@
 package com.linkedin.venice.meta;
 
 public interface StoreCleaner {
-  void deleteOneStoreVersion(String clusterName, String storeName, int versionNumber);
+  void deleteOneStoreVersion(
+      String clusterName,
+      String storeName,
+      int versionNumber,
+      boolean isForcedDelete,
+      boolean deleteDueToError);
 
   void retireOldStoreVersions(
       String clusterName,

internal/venice-common/src/main/java/com/linkedin/venice/participant/protocol/enums/PushJobKillTrigger.java

@@ -0,0 +1,72 @@
+package com.linkedin.venice.participant.protocol.enums;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+/**
+ * Represents the trigger that caused an offline push job to be terminated.
+ */
+public enum PushJobKillTrigger {
+  // User/system-initiated actions
+  USER_REQUEST(0), // Push job was killed due to explicit user request
+  VERSION_RETIREMENT(1), // Push job was killed due to version retirement
+
+  // System-detected violations or preemption
+  SLA_VIOLATION(2), // Push job violated SLA (e.g., took too long)
+  PREEMPTED_BY_FULL_PUSH(3), // Repush was preempted by a new full/batch push
+
+  // Failures during ingestion or version setup
+  INGESTION_FAILURE(4), // Push job failed due to ingestion errors
+  VERSION_CREATION_FAILURE(5), // Push job failed due to version creation errors
+  PUSH_JOB_FAILED(6), // Generic failure during push job
+
+  // Cleanup and maintenance triggers
+  LINGERING_VERSION_TOPIC(7), // Push job was killed due to lingering version topic
+
+  // Catch-all
+  UNKNOWN(8); // Unknown reason
+
+  private final int code;
+
+  private static final Map<Integer, PushJobKillTrigger> CODE_MAP = new HashMap<>(8);
+  private static final Map<String, PushJobKillTrigger> NAME_MAP = new HashMap<>(8);
+
+  static {
+    for (PushJobKillTrigger trigger: values()) {
+      CODE_MAP.put(trigger.code, trigger);
+      NAME_MAP.put(trigger.name().toLowerCase(), trigger);
+    }
+  }
+
+  PushJobKillTrigger(int code) {
+    this.code = code;
+  }
+
+  public int getCode() {
+    return code;
+  }
+
+  /**
+   * Returns the enum constant for the given code.
+   *
+   * @param code the integer code
+   * @return the corresponding KillPushJobTrigger, or null if not found
+   */
+  public static PushJobKillTrigger fromCode(int code) {
+    return CODE_MAP.get(code);
+  }
+
+  /**
+   * Returns the enum constant for the given name (case-insensitive).
+   *
+   * @param name the name of the enum constant
+   * @return the corresponding KillPushJobTrigger, or null if not found
+   */
+  public static PushJobKillTrigger fromString(String name) {
+    if (name == null) {
+      return null;
+    }
+    return NAME_MAP.get(name.toLowerCase());
+  }
+}

internal/venice-common/src/main/java/com/linkedin/venice/serialization/avro/AvroProtocolDefinition.java

@@ -77,7 +77,7 @@ public enum AvroProtocolDefinition {
    *
    * TODO: Move AdminOperation to venice-common module so that we can properly reference it here.
    */
-  ADMIN_OPERATION(94, SpecificData.get().getSchema(ByteBuffer.class), "AdminOperation"),
+  ADMIN_OPERATION(96, SpecificData.get().getSchema(ByteBuffer.class), "AdminOperation"),
 
   /**
    * Single chunk of a large multi-chunk value. Just a bunch of bytes.
@@ -168,7 +168,7 @@ public enum AvroProtocolDefinition {
   /**
    * Value schema for participant system stores.
    */
-  PARTICIPANT_MESSAGE_SYSTEM_STORE_VALUE(1, ParticipantMessageValue.class),
+  PARTICIPANT_MESSAGE_SYSTEM_STORE_VALUE(2, ParticipantMessageValue.class),
 
   /**
    * Response record for admin request.