Fix IndexOutOfBoundsException invoking expansion service (#34428)

Author: Abacn

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

@@ -580,10 +580,8 @@ private Map<String, TransformProvider> loadRegisteredTransforms() {
 
   @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 @@ private Map<String, TransformProvider> loadRegisteredTransforms() {
                       }
                     }));
 
-    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 @@ private Map<String, TransformProvider> loadRegisteredTransforms() {
                 ? transformProvider
                 : ExpansionServiceSchemaTransformProvider.of();
       } else {
-        throw new UnsupportedOperationException(
-            "Unknown urn: " + request.getTransform().getSpec().getUrn());
+        throw new UnsupportedOperationException("Unknown urn: " + urn);
       }
     }
 
@@ -814,6 +809,10 @@ public void close() throws Exception {
   }
 
   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 static void main(String[] args) throws Exception {
     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();