From 86f375efb7f5425650139769e1873c516cdb956d Mon Sep 17 00:00:00 2001
From: Ben Sidhom <sidhom@google.com>
Date: Fri, 13 Apr 2018 11:07:24 -0700
Subject: [PATCH] [BEAM-4069] Gracefully deserialize empty options structs

---
 .../PipelineOptionsTranslationTest.java       | 21 +++++++++++++++++++
 .../sdk/options/ProxyInvocationHandler.java   | 16 ++++++++------
 2 files changed, 31 insertions(+), 6 deletions(-)

diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PipelineOptionsTranslationTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PipelineOptionsTranslationTest.java
index eb59bac95a10..24fd3811d34f 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PipelineOptionsTranslationTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/PipelineOptionsTranslationTest.java
@@ -20,12 +20,15 @@
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertThat;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.common.collect.ImmutableList;
+import com.google.protobuf.NullValue;
 import com.google.protobuf.Struct;
+import com.google.protobuf.Value;
 import org.apache.beam.sdk.options.ApplicationNameOptions;
 import org.apache.beam.sdk.options.Default;
 import org.apache.beam.sdk.options.PipelineOptions;
@@ -116,6 +119,24 @@ public void defaultsRestored() throws Exception {
 
       assertThat(deserialized.as(TestDefaultOptions.class).getDefault(), equalTo(19));
     }
+
+    @Test
+    public void emptyStructDeserializes() throws Exception {
+      Struct serialized = Struct.getDefaultInstance();
+      PipelineOptions deserialized = PipelineOptionsTranslation.fromProto(serialized);
+
+      assertThat(deserialized, notNullValue());
+    }
+
+    @Test
+    public void structWithNullOptionsDeserializes() throws Exception {
+      Struct serialized = Struct.newBuilder()
+          .putFields("options", Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+          .build();
+      PipelineOptions deserialized = PipelineOptionsTranslation.fromProto(serialized);
+
+      assertThat(deserialized, notNullValue());
+    }
   }
 
   /** {@link PipelineOptions} with an unserializable option. */
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
index a127fd15e256..ccbe2ff50f8f 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ProxyInvocationHandler.java
@@ -732,15 +732,19 @@ static class Deserializer extends JsonDeserializer<PipelineOptions> {
     @Override
     public PipelineOptions deserialize(JsonParser jp, DeserializationContext ctxt)
         throws IOException, JsonProcessingException {
-      ObjectNode objectNode = (ObjectNode) jp.readValueAsTree();
-      ObjectNode optionsNode = (ObjectNode) objectNode.get("options");
+      ObjectNode objectNode = jp.readValueAsTree();
+      JsonNode rawOptionsNode = objectNode.get("options");
 
       Map<String, JsonNode> fields = Maps.newHashMap();
-      for (Iterator<Map.Entry<String, JsonNode>> iterator = optionsNode.fields();
-          iterator.hasNext(); ) {
-        Map.Entry<String, JsonNode> field = iterator.next();
-        fields.put(field.getKey(), field.getValue());
+      if (rawOptionsNode != null && !rawOptionsNode.isNull()) {
+        ObjectNode optionsNode = (ObjectNode) rawOptionsNode;
+        for (Iterator<Map.Entry<String, JsonNode>> iterator = optionsNode.fields();
+            iterator != null && iterator.hasNext(); ) {
+          Map.Entry<String, JsonNode> field = iterator.next();
+          fields.put(field.getKey(), field.getValue());
+        }
       }
+
       PipelineOptions options =
           new ProxyInvocationHandler(Maps.newHashMap(), fields).as(PipelineOptions.class);
       ValueProvider.RuntimeValueProvider.setRuntimeOptions(options);