(beam) branch master updated: Fix IndexOutOfBoundsException invoking expansion service (#34428)

[yh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 1540776dc30 Fix IndexOutOfBoundsException invoking expansion service (#34428)
1540776dc30 is described below

commit 1540776dc3031aba15b229732ec9d552d4372d9f
Author: Yi Hu <ya...@google.com>
AuthorDate: Thu Apr 24 18:40:17 2025 -0400

    Fix IndexOutOfBoundsException invoking expansion service (#34428)
---
 .../beam/sdk/expansion/service/ExpansionService.java  | 19 +++++++++++--------
 1 file changed, 11 insertions(+), 8 deletions(-)

diff --git a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java b/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java
index 4b0457320dd..2bd45067918 100644
--- a/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java
+++ b/sdks/java/expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java
@@ -580,10 +580,8 @@ public class ExpansionService extends ExpansionServiceGrpc.ExpansionServiceImplB
 
   @VisibleForTesting
   /*package*/ ExpansionApi.ExpansionResponse expand(ExpansionApi.ExpansionRequest request) {
-    LOG.info(
-        "Expanding '{}' with URN '{}'",
-        request.getTransform().getUniqueName(),
-        request.getTransform().getSpec().getUrn());
+    final String urn = request.getTransform().getSpec().getUrn();
+    LOG.info("Expanding '{}' with URN '{}'", request.getTransform().getUniqueName(), urn);
     LOG.debug("Full transform: {}", request.getTransform());
     Set<String> existingTransformIds = request.getComponents().getTransformsMap().keySet();
 
@@ -625,8 +623,6 @@ public class ExpansionService extends ExpansionServiceGrpc.ExpansionServiceImplB
                       }
                     }));
 
-    String urn = request.getTransform().getSpec().getUrn();
-
     TransformProvider transformProvider = getRegisteredTransforms().get(urn);
     if (transformProvider == null) {
       if (getUrn(ExpansionMethods.Enum.JAVA_CLASS_LOOKUP).equals(urn)) {
@@ -649,8 +645,7 @@ public class ExpansionService extends ExpansionServiceGrpc.ExpansionServiceImplB
                 ? transformProvider
                 : ExpansionServiceSchemaTransformProvider.of();
       } else {
-        throw new UnsupportedOperationException(
-            "Unknown urn: " + request.getTransform().getSpec().getUrn());
+        throw new UnsupportedOperationException("Unknown urn: " + urn);
       }
     }
 
@@ -814,6 +809,10 @@ public class ExpansionService extends ExpansionServiceGrpc.ExpansionServiceImplB
   }
 
   public static void main(String[] args) throws Exception {
+    if (args.length < 1) {
+      printUsage();
+      System.exit(1);
+    }
     int port = Integer.parseInt(args[0]);
     System.out.println("Starting expansion service at localhost:" + port);
 
@@ -874,6 +873,10 @@ public class ExpansionService extends ExpansionServiceGrpc.ExpansionServiceImplB
     server.awaitTermination();
   }
 
+  private static void printUsage() {
+    System.err.println("Usage: java -jar <expansion-service-jar> port [pipeline-options]");
+  }
+
   private static class NotRunnableRunner extends PipelineRunner<PipelineResult> {
     public static NotRunnableRunner fromOptions(PipelineOptions opts) {
       return new NotRunnableRunner();