diff --git a/clients/venice-push-job/src/main/java/com/linkedin/venice/spark/utils/RmdPushUtils.java b/clients/venice-push-job/src/main/java/com/linkedin/venice/spark/utils/RmdPushUtils.java index 17edbdcc472..5d79d4b4787 100644 --- a/clients/venice-push-job/src/main/java/com/linkedin/venice/spark/utils/RmdPushUtils.java +++ b/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) { diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/utils/TestRmdPushUtils.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/utils/TestRmdPushUtils.java index 841a52f800e..4acfab14207 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/utils/TestRmdPushUtils.java +++ b/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";