From 2552c6fe78ddeafc30798ed3896475cf445fcf75 Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Tue, 30 Jun 2026 10:37:13 -0700 Subject: [PATCH 1/2] [vpj] Throw a clear error instead of NPE on a missing timestamp.field RmdPushUtils.getInputRmdSchema looked up the configured timestamp.field (RMD_FIELD_PROP) via inputDataSchema.getField(rmdField).schema() with no null check. Avro's Schema.getField returns null when the field is not a top-level field of the input record (for example when it is nested inside the value, misnamed, or absent), so a batch push that sets timestamp.field but does not expose it at the input record root failed with an opaque NullPointerException that gave no hint at the real cause. Null-check the looked-up field and throw VeniceSchemaFieldNotFoundException with an actionable message that names the configured timestamp.field and states it must be a top-level field of the input record. This mirrors the existing behavior of AbstractAvroRecordReader.getField on the MR path, which already fails with VeniceSchemaFieldNotFoundException for the same case. Add a unit test asserting the clear exception (not an NPE) is thrown when the configured field is absent from the input schema. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../linkedin/venice/spark/utils/RmdPushUtils.java | 12 +++++++++++- .../venice/hadoop/utils/TestRmdPushUtils.java | 15 +++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) 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..c137799b858 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,7 @@ package com.linkedin.venice.spark.utils; 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 +27,16 @@ 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(); + 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..68dccbe40aa 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 @@ -8,6 +8,7 @@ import static org.testng.Assert.assertTrue; 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 +36,20 @@ 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 public void testContainsLogicalTimestamp() { final String rmdField = "rmd"; From 29b767fb6b5cad17388fb6f108f0355f6740c453 Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Tue, 30 Jun 2026 11:35:09 -0700 Subject: [PATCH 2/2] [vpj] Also guard null inputDataSchema in getInputRmdSchema Address review feedback: getInputRmdSchema dereferenced pushJobSetting.inputDataSchema before looking up the field, so a push that configures timestamp.field on an input that never populates inputDataSchema (e.g. VSON inputs, whose DefaultInputDataInfoProvider branch sets rmdField without setting inputDataSchema) would still fail with an opaque NullPointerException. Add an explicit null-schema guard that throws VeniceException naming the configured timestamp.field, plus a unit test. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../com/linkedin/venice/spark/utils/RmdPushUtils.java | 7 +++++++ .../venice/hadoop/utils/TestRmdPushUtils.java | 11 +++++++++++ 2 files changed, 18 insertions(+) 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 c137799b858..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,5 +1,6 @@ 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; @@ -27,6 +28,12 @@ public static Schema getInputRmdSchema(PushJobSetting pushJobSetting) { throw new IllegalArgumentException( "Push job setting missing rmd field. Please set the rmd field in the job properties."); } + 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( 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 68dccbe40aa..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,6 +7,7 @@ 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; @@ -50,6 +51,16 @@ public void testGetInputRmdSchemaWhenRmdFieldNotInInputSchema() { 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";