Updates the Java Expansion Service container to support gRPC ALTS aut…

…hentication (#31352)

* Updates the Java Expansion Service container to support Google ALTS authentication

* Fix spotless

* Updates the description

sdks/java/expansion-service/container/boot.go

@@ -44,6 +44,7 @@ var (
 	port             = flag.Int("port", 0, "Port for the expansion service (required)")
 	dependencies_dir = flag.String("dependencies_dir", "", "A directory containing the set of jar files to load transforms from (required)")
 	config_file      = flag.String("config_file", "", "Expansion service config YAML file. (required)")
+	use_alts      = flag.Bool("use_alts", false, "Starts an Expansion Service with support for gRPC ALTS authentication")
 )
 
 const entrypoint = "org.apache.beam.sdk.expansion.service.ExpansionService"
@@ -90,6 +91,9 @@ func main() {
 	if *config_file != "" {
 		args = append(args, fmt.Sprintf("--expansionServiceConfigFile=%s", *config_file))
 	}
+	if *use_alts {
+		args = append(args, "--useAltsServer=true")
+	}
 
 	log.Printf("Executing: java %v", strings.Join(args, " "))
 	log.Fatalf("Java exited: %v", execx.Execute("java", args...))

sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

@@ -88,6 +88,7 @@
 import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.Server;
 import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ServerBuilder;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.alts.AltsServerBuilder;
 import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.stub.StreamObserver;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.CaseFormat;
@@ -823,8 +824,15 @@ public static void main(String[] args) throws Exception {
       System.out.println("\nDid not find any registered transforms or SchemaTransforms.\n");
     }
 
+    boolean useAlts = options.as(ExpansionServiceOptions.class).getUseAltsServer();
     ServerBuilder serverBuilder =
-        ServerBuilder.forPort(port).addService(service).addService(new ArtifactRetrievalService());
+        useAlts ? AltsServerBuilder.forPort(port) : ServerBuilder.forPort(port);
+
+    if (useAlts) {
+      LOG.info("Running with gRPC ALTS authentication.");
+    }
+
+    serverBuilder.addService(service).addService(new ArtifactRetrievalService());
     if (options.as(ExpansionServiceOptions.class).getAlsoStartLoopbackWorker()) {
       serverBuilder.addService(new ExternalWorkerService(options));
     }

sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionServiceOptions.java

@@ -58,6 +58,12 @@ public interface ExpansionServiceOptions extends PipelineOptions {
 
   void setExpansionServiceConfig(ExpansionServiceConfig configFile);
 
+  @Description("Starts an Expansion Service with support for gRPC ALTS authentication.")
+  @Default.Boolean(false)
+  boolean getUseAltsServer();
+
+  void setUseAltsServer(boolean useAltsServer);
+
   /**
    * Loads the allow list from {@link #getJavaClassLookupAllowlistFile}, defaulting to an empty
    * {@link JavaClassLookupTransformProvider.AllowList}.

sdks/java/io/expansion-service/build.gradle

@@ -27,6 +27,10 @@ applyJavaNature(
   shadowClosure: {},
 )
 
+shadowJar {
+  mergeServiceFiles()
+}
+
 description = "Apache Beam :: SDKs :: Java :: IO :: Expansion Service"
 ext.summary = "Expansion service serving several Java IOs"