diff --git a/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json b/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json index b08e2bb81150..1fdefde8600e 100644 --- a/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json +++ b/.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Flink.json @@ -1,5 +1,5 @@ { "https://github.com/apache/beam/pull/32648": "testing addition of Flink 1.19 support", "https://github.com/apache/beam/pull/34830": "testing", - "trigger-2026-04-04": "portable_runner expand_sdf opt-in" + "trigger-2026-05-18": "Fix deserializeNode NPE" } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java index ac76a57b6b07..3175ee56935e 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java @@ -1746,7 +1746,7 @@ private static JsonDeserializer computeDeserializerForMethod(Method meth TypeDeserializer typeDeserializer = context.getFactory().findTypeDeserializer(context.getConfig(), prop.getType()); - if (typeDeserializer != null) { + if (typeDeserializer != null && jsonDeserializer != null) { jsonDeserializer = new TypeWrappedDeserializer(typeDeserializer, jsonDeserializer); } @@ -1796,15 +1796,33 @@ private static JsonDeserializer getDeserializerForMethod(Method method) .orNull(); } + /** + * Returns true when the getter declares custom Jackson serde, per {@link PipelineOptions} + * validation ({@code @JsonSerialize} and {@code @JsonDeserialize} must appear together). + */ + private static boolean usesPropertyLevelJacksonSerde(Method method) { + return method.isAnnotationPresent(JsonSerialize.class) + && method.isAnnotationPresent(JsonDeserialize.class); + } + static Object deserializeNode(JsonNode node, Method method) throws IOException { if (node.isNull()) { return null; } - JsonParser parser = new TreeTraversingParser(node, MAPPER); - parser.nextToken(); + if (!usesPropertyLevelJacksonSerde(method)) { + JavaType javaType = MAPPER.constructType(method.getGenericReturnType()); + return MAPPER.treeToValue(node, javaType); + } JsonDeserializer jsonDeserializer = getDeserializerForMethod(method); + if (jsonDeserializer == null) { + JavaType javaType = MAPPER.constructType(method.getGenericReturnType()); + return MAPPER.treeToValue(node, javaType); + } + + JsonParser parser = new TreeTraversingParser(node, MAPPER); + parser.nextToken(); return jsonDeserializer.deserialize(parser, DESERIALIZATION_CONTEXT.copy()); } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java index 2755ade271dd..adb3f0e4dc0f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java @@ -74,6 +74,7 @@ import org.apache.beam.sdk.testing.ExpectedLogs; import org.apache.beam.sdk.testing.InterceptingUrlClassLoader; import org.apache.beam.sdk.testing.RestoreSystemProperties; +import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.util.common.ReflectHelpers; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ArrayListMultimap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Collections2; @@ -1130,6 +1131,23 @@ public void testPolymorphicType() { assertEquals(PolymorphicTypeTwo.class, options.getObjectValue().get().getClass()); } + /** + * Regression test for display-data population after JSON round-trip. {@link DisplayData#from} + * exercises {@link PipelineOptionsFactory#deserializeNode} for jsonOptions (treeToValue path). + */ + @Test + public void testDisplayDataFromDeserializedPolymorphicOption() throws Exception { + PolymorphicTypes options = + PipelineOptionsFactory.fromArgs("--object={\"key\":\"value\",\"@type\":\"one\"}") + .as(PolymorphicTypes.class); + ObjectMapper mapper = + new ObjectMapper() + .registerModules(ObjectMapper.findModules(ReflectHelpers.findClassLoader())); + PipelineOptions deserialized = + mapper.readValue(mapper.writeValueAsString(options), PipelineOptions.class); + DisplayData.from(deserialized); + } + @Test public void testMissingArgument() { String[] args = new String[] {};