Set default options from boot.go; move PipelineOpt to SdkHarnessOptions

Author: clairemcginty

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java

@@ -18,7 +18,6 @@
 package org.apache.beam.runners.dataflow.options;
 
 import com.google.api.services.dataflow.Dataflow;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.runners.dataflow.DataflowRunner;
@@ -267,30 +266,4 @@ public String create(PipelineOptions options) {
   List<String> getJdkAddOpenModules();
 
   void setJdkAddOpenModules(List<String> options);
-
-  class AvroSerializableClassesFactory implements DefaultValueFactory<List<String>> {
-    @Override
-    public List<String> create(PipelineOptions options) {
-      return Arrays.asList(
-          "java.math.BigDecimal",
-          "java.math.BigInteger",
-          "java.net.URI",
-          "java.net.URL",
-          "java.io.File",
-          "java.lang.Integer");
-    }
-  }
-
-  /**
-   * The Avro spec supports the `java-class` schema annotation, which allows fields to be serialized
-   * and deserialized via their toString/String constructor. As of Avro 1.11.4+, allowed Java
-   * classes must be explicitly specified via the jvm option. The comma-separated String value of
-   * this pipeline option will be passed to the Dataflow worker via the
-   * -Dorg.apache.avro.SERIALIZABLE_CLASSES jvm option.
-   */
-  @Description("Serializable classes required by java-class props in Avro 1.11.4+")
-  @Default.InstanceFactory(AvroSerializableClassesFactory.class)
-  List<String> getAvroSerializableClasses();
-
-  void setAvroSerializableClasses(List<String> options);
 }

runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptionsTest.java

@@ -29,7 +29,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.TimeoutException;
 import org.apache.beam.sdk.extensions.gcp.storage.NoopPathValidator;
@@ -253,28 +252,6 @@ public void testDefaultGcpRegionFromGcloud() {
     }
   }
 
-  @Test
-  public void testDefaultAvroSerializableClasses() {
-    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
-    assertEquals(
-        Arrays.asList(
-            "java.math.BigDecimal",
-            "java.math.BigInteger",
-            "java.net.URI",
-            "java.net.URL",
-            "java.io.File",
-            "java.lang.Integer"),
-        options.getAvroSerializableClasses());
-  }
-
-  @Test
-  public void testOverriddenAvroSerializableClasses() {
-    final List<String> opts = Arrays.asList("foo", "bar");
-    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
-    options.setAvroSerializableClasses(opts);
-    assertEquals(opts, options.getAvroSerializableClasses());
-  }
-
   /**
    * If gcloud gets stuck, test that {@link DefaultGcpRegionFactory#getRegionFromGcloudCli(long)}
    * times out instead of blocking forever.

sdks/java/container/boot.go

@@ -277,12 +277,23 @@ func main() {
 			}
 		}
 		// Add trusted Avro serializable classes
+		var serializableClassesList []string
 		if serializableClasses, ok := pipelineOptions.GetStructValue().GetFields()["avroSerializableClasses"]; ok {
-			var serializableClassesSlice []string
 			for _, cls := range serializableClasses.GetListValue().GetValues() {
-				serializableClassesSlice = append(serializableClassesSlice, cls.GetStringValue())
+				serializableClassesList = append(serializableClassesList, cls.GetStringValue())
 			}
-			args = append(args, "-Dorg.apache.avro.SERIALIZABLE_CLASSES="+strings.Join(serializableClassesSlice, ","))
+		} else {
+			serializableClassesList = []string{
+				"java.math.BigDecimal",
+				"java.math.BigInteger",
+				"java.net.URI",
+				"java.net.URL",
+				"java.io.File",
+				"java.lang.Integer",
+			}
+		}
+		if len(serializableClassesList) > 0 {
+			args = append(args, "-Dorg.apache.avro.SERIALIZABLE_CLASSES="+strings.Join(serializableClassesList, ","))
 		}
 	}
 	// Automatically open modules for Java 11+

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

@@ -440,4 +440,16 @@ public Duration create(PipelineOptions options) {
   int getElementProcessingTimeoutMinutes();
 
   void setElementProcessingTimeoutMinutes(int value);
+
+  /**
+   * The Avro spec supports the `java-class` schema annotation, which allows fields to be serialized
+   * and deserialized via their toString/String constructor. As of Avro 1.11.4+, allowed Java
+   * classes must be explicitly specified via the jvm option. The comma-separated String value of
+   * this pipeline option will be passed to the Dataflow worker via the
+   * -Dorg.apache.avro.SERIALIZABLE_CLASSES jvm option.
+   */
+  @Description("Serializable classes required by java-class props in Avro 1.11.4+")
+  List<String> getAvroSerializableClasses();
+
+  void setAvroSerializableClasses(List<String> options);
 }