Fix PipelineOptions display-data NPE

.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"
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java

@@ -1746,7 +1746,7 @@ private static JsonDeserializer<Object> 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<Object> 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<Object> 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());
   }
 

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[] {};