[vpj] Address review feedback for kill-check monitor

- Reset pushJobKilledByController flag at start of runJobWithKillDetection()
- Add @VisibleForTesting to package-private test-accessible methods
- Add early return guard to prevent repeated kill logging/calls
- Clarify response.isError() (HTTP) vs status.isError() (push status)
- Escalate HTTP query failures from WARN to ERROR
- Use ExecutionStatus.ERROR.toString() instead of string literal
- Verify queryOverallJobStatus was invoked in kill detection test
- Replace fragile skipVPJValidation + re-override with explicit stubs

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pthirun

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

@@ -83,6 +83,7 @@
 import static com.linkedin.venice.vpj.VenicePushJobConstants.VENICE_DISCOVER_URL_PROP;
 import static com.linkedin.venice.vpj.VenicePushJobConstants.VENICE_STORE_NAME_PROP;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import com.linkedin.d2.balancer.D2Client;
 import com.linkedin.venice.PushJobCheckpoints;
@@ -1051,7 +1052,9 @@ private void setupJobTimeoutMonitor() {
    * after data writing completes but before the monitor is cancelled. This is correct behavior — if the
    * push was killed, any data written is wasted, and we should still fail the job.
    */
+  @VisibleForTesting
   void runJobWithKillDetection() {
+    pushJobKilledByController = false;
     startPushJobKillCheckMonitor();
     try {
       runJobAndUpdateStatus();
@@ -1066,6 +1069,7 @@ void runJobWithKillDetection() {
    * This runs during the data writing phase to detect early kills (e.g., when a user push supersedes
    * a repush) and abort the data writer job promptly instead of wasting resources.
    */
+  @VisibleForTesting
   void startPushJobKillCheckMonitor() {
     String topicToMonitor = getTopicToMonitor(pushJobSetting);
     long intervalMs = pushJobSetting.pollJobStatusIntervalMs;
@@ -1076,12 +1080,17 @@ void startPushJobKillCheckMonitor() {
         intervalMs);
     pushJobKillCheckScheduledFuture = timeoutExecutor.scheduleWithFixedDelay(() -> {
       try {
+        if (pushJobKilledByController) {
+          return;
+        }
         JobStatusQueryResponse response = ControllerClient.retryableRequest(
             controllerClient,
             pushJobSetting.controllerStatusPollRetries,
             client -> client.queryOverallJobStatus(topicToMonitor, Optional.empty(), null, false));
+        // response.isError() indicates an HTTP/transport error (failed to reach the controller),
+        // NOT that the push status is ERROR. Push status is checked separately below via status.isError().
         if (response.isError()) {
-          LOGGER.warn(
+          LOGGER.error(
               "Kill check monitor could not query job status for store: {}, version: {}. Error: {}",
               pushJobSetting.storeName,
               pushJobSetting.version,
@@ -1104,6 +1113,7 @@ void startPushJobKillCheckMonitor() {
     }, intervalMs, intervalMs, TimeUnit.MILLISECONDS);
   }
 
+  @VisibleForTesting
   void stopPushJobKillCheckMonitor() {
     if (pushJobKillCheckScheduledFuture != null) {
       pushJobKillCheckScheduledFuture.cancel(false);

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

@@ -1933,7 +1933,7 @@ public void testPushJobKilledDuringDataWriting(Class<? extends DataWriterCompute
 
     // Simulate controller returning ERROR status (push was killed)
     JobStatusQueryResponse killResponse = mock(JobStatusQueryResponse.class);
-    doReturn("ERROR").when(killResponse).getStatus();
+    doReturn(ExecutionStatus.ERROR.toString()).when(killResponse).getStatus();
     doReturn(false).when(killResponse).isError();
     doReturn(killResponse).when(client).queryOverallJobStatus(anyString(), any(), any(), anyBoolean());
 
@@ -1972,12 +1972,27 @@ public void testPushJobKilledDuringDataWriting(Class<? extends DataWriterCompute
         return null;
       }).when(dataWriterJob).kill();
 
-      // N.B.: skipVPJValidation stubs runJobAndUpdateStatus() (among others), so we must re-override
-      // it afterward to let the real kill-detection flow execute. If skipVPJValidation's stubs change,
-      // this re-override may need updating.
-      skipVPJValidation(pushJob);
-      doCallRealMethod().when(pushJob).runJobWithKillDetection();
-      doCallRealMethod().when(pushJob).runJobAndUpdateStatus();
+      // Stub only the validation methods from skipVPJValidation that this test needs bypassed.
+      // We intentionally avoid skipVPJValidation() because it also stubs runJobAndUpdateStatus(),
+      // which we need to run for kill-detection to work.
+      doAnswer(invocation -> {
+        VeniceProperties properties = pushJob.getJobProperties();
+        PushJobSetting pjs = pushJob.getPushJobSetting();
+        if (!pjs.isSourceKafka) {
+          Schema schema = AvroSchemaParseUtils.parseSchemaFromJSONLooseValidation(SIMPLE_FILE_SCHEMA_STR);
+          pjs.keyField = properties.getString(KEY_FIELD_PROP, DEFAULT_KEY_FIELD_PROP);
+          pjs.valueField = properties.getString(VALUE_FIELD_PROP, DEFAULT_VALUE_FIELD_PROP);
+          pjs.inputDataSchema = schema;
+          pjs.valueSchema = schema.getField(pjs.valueField).schema();
+          pjs.inputDataSchemaString = SIMPLE_FILE_SCHEMA_STR;
+          pjs.keySchema = pjs.inputDataSchema.getField(pjs.keyField).schema();
+          pjs.keySchemaString = pjs.keySchema.toString();
+          pjs.valueSchemaString = pjs.valueSchema.toString();
+        }
+        return getMockInputDataInfoProvider();
+      }).when(pushJob).getInputDataInfoProvider();
+      doNothing().when(pushJob).validateKeySchema(any());
+      doNothing().when(pushJob).validateAndRetrieveValueSchemas(any(), any(), anyBoolean());
 
       try {
         pushJob.run();
@@ -1992,6 +2007,7 @@ public void testPushJobKilledDuringDataWriting(Class<? extends DataWriterCompute
       assertEquals(dataWriterRunningLatch.getCount(), 0, "Data writer job should have started");
       assertEquals(dataWriterKilledLatch.getCount(), 0, "Data writer job should have been killed");
       verify(dataWriterJob, times(1)).kill();
+      verify(client, atLeastOnce()).queryOverallJobStatus(anyString(), any(), any(), anyBoolean());
     }
   }