Allow users to specify trusted Avro serializable classes to Dataflow worker (#36809)

* [Proposal] Allow users to specify trusted Avro serializable classes to Dataflow worker

* Fixup boot.go

* Add default factory; add tests

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

* Add check for empty list

Author: clairemcginty

sdks/java/container/boot.go

@@ -276,6 +276,28 @@ func main() {
 				args = append(args, "--add-modules="+module.GetStringValue())
 			}
 		}
+		// Add trusted Avro serializable classes
+		var serializableClassesList []string
+		if serializableClasses, ok := pipelineOptions.GetStructValue().GetFields()["avroSerializableClasses"]; ok {
+			for _, cls := range serializableClasses.GetListValue().GetValues() {
+				// User can specify an empty list, which is serialized as a single, blank value
+				if cls.GetStringValue() != "" {
+					serializableClassesList = append(serializableClassesList, cls.GetStringValue())
+				}
+			}
+		} 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+
 	openModuleAgentJar := "/opt/apache/beam/jars/open-module-agent.jar"

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);
 }