[vpj] Throw a clear error instead of NPE on a missing timestamp.field

clients/venice-push-job/src/main/java/com/linkedin/venice/spark/utils/RmdPushUtils.java

@@ -1,6 +1,8 @@
 package com.linkedin.venice.spark.utils;
 
+import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.hadoop.PushJobSetting;
+import com.linkedin.venice.hadoop.exceptions.VeniceSchemaFieldNotFoundException;
 import com.linkedin.venice.serializer.FastSerializerDeserializerFactory;
 import com.linkedin.venice.serializer.RecordDeserializer;
 import com.linkedin.venice.serializer.RecordSerializer;
@@ -26,7 +28,22 @@ public static Schema getInputRmdSchema(PushJobSetting pushJobSetting) {
       throw new IllegalArgumentException(
           "Push job setting missing rmd field. Please set the rmd field in the job properties.");
     }
-    return pushJobSetting.inputDataSchema.getField(pushJobSetting.rmdField).schema();
+    if (pushJobSetting.inputDataSchema == null) {
+      throw new VeniceException(
+          "The configured timestamp.field '" + pushJobSetting.rmdField
+              + "' cannot be resolved because the push input record schema is not available. A top-level Avro "
+              + "input record schema is required to use timestamp.field (for example, VSON inputs do not populate it).");
+    }
+    Schema.Field rmdField = pushJobSetting.inputDataSchema.getField(pushJobSetting.rmdField);
+    if (rmdField == null) {
+      throw new VeniceSchemaFieldNotFoundException(
+          pushJobSetting.rmdField,
+          "Could not find the configured timestamp.field '" + pushJobSetting.rmdField
+              + "' at the top level of the push input record schema. The timestamp field must be a top-level field "
+              + "of the input record (a sibling of the key and value fields), not nested inside the value. "
+              + "Input record schema: " + pushJobSetting.inputDataSchema);
+    }
+    return rmdField.schema();
   }
 
   public static boolean containsLogicalTimestamp(PushJobSetting pushJobSetting) {

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

@@ -7,7 +7,9 @@
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
+import com.linkedin.venice.exceptions.VeniceException;
 import com.linkedin.venice.hadoop.PushJobSetting;
+import com.linkedin.venice.hadoop.exceptions.VeniceSchemaFieldNotFoundException;
 import com.linkedin.venice.spark.utils.RmdPushUtils;
 import org.apache.avro.Schema;
 import org.testng.annotations.Test;
@@ -35,6 +37,30 @@ public void testGetInputRmdSchemaWithNoRmdField() {
     RmdPushUtils.getInputRmdSchema(pushJobSetting);
   }
 
+  @Test(expectedExceptions = VeniceSchemaFieldNotFoundException.class)
+  public void testGetInputRmdSchemaWhenRmdFieldNotInInputSchema() {
+    // timestamp.field is configured, but the named field is absent from the top level of the input record schema
+    // (e.g. it is nested inside the value). getField returns null, so the call must fail with a clear error, not an
+    // NPE.
+    final String rmdField = "createdTimeEpoch";
+    Schema mockSchema = mock(Schema.class);
+    when(mockSchema.getField(eq(rmdField))).thenReturn(null);
+    PushJobSetting pushJobSetting = new PushJobSetting();
+    pushJobSetting.rmdField = rmdField;
+    pushJobSetting.inputDataSchema = mockSchema;
+    RmdPushUtils.getInputRmdSchema(pushJobSetting);
+  }
+
+  @Test(expectedExceptions = VeniceException.class)
+  public void testGetInputRmdSchemaWhenInputDataSchemaIsNull() {
+    // timestamp.field is configured but the input record schema was never populated (e.g. a VSON input). The
+    // configured-but-unavailable schema must fail with a clear error, not an NPE.
+    PushJobSetting pushJobSetting = new PushJobSetting();
+    pushJobSetting.rmdField = "createdTimeEpoch";
+    pushJobSetting.inputDataSchema = null;
+    RmdPushUtils.getInputRmdSchema(pushJobSetting);
+  }
+
   @Test
   public void testContainsLogicalTimestamp() {
     final String rmdField = "rmd";