Fix for FQFN flag (#1528)

* move fqfn flag into proper command base

* Extend FunctionCommand from BaseCommand rather than NamespaceCommand

* clean up input topics function

* Add special processArguments function for FunctionCommand

* Match wording of error output across commands

* Add checks for tenant, namespace, and function name

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.net.MalformedURLException;
+import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -103,9 +104,6 @@ void processArguments() throws Exception {}
      */
     @Getter
     abstract class NamespaceCommand extends BaseCommand {
-        @Parameter(names = "--fqfn", description = "The Fully Qualified Function Name (FQFN) for the function", required = false)
-        protected String fqfn;
-
         @Parameter(names = "--tenant", description = "The function's tenant", required = true)
         protected String tenant;
 
@@ -117,9 +115,47 @@ abstract class NamespaceCommand extends BaseCommand {
      * Function level command
      */
     @Getter
-    abstract class FunctionCommand extends NamespaceCommand {
-        @Parameter(names = "--name", description = "The function's name", required = true)
+    abstract class FunctionCommand extends BaseCommand {
+        @Parameter(names = "--fqfn", description = "The Fully Qualified Function Name (FQFN) for the function")
+        protected String fqfn;
+
+        @Parameter(names = "--tenant", description = "The function's tenant")
+        protected String tenant;
+
+        @Parameter(names = "--namespace", description = "The function's namespace")
+        protected String namespace;
+
+        @Parameter(names = "--name", description = "The function's name")
         protected String functionName;
+
+        @Override
+        void processArguments() throws Exception {
+            super.processArguments();
+
+            boolean usesSetters = (null != tenant || null != namespace || null != functionName);
+            boolean usesFqfn = (null != fqfn);
+
+            // Throw an exception if --fqfn is set alongside any combination of --tenant, --namespace, and --name
+            if (usesFqfn && usesSetters) {
+                throw new RuntimeException(
+                        "You must specify either a Fully Qualified Function Name (FQFN) or tenant, namespace, and function name");
+            } else if (usesFqfn) {
+                // If the --fqfn flag is used, parse tenant, namespace, and name using that flag
+                String[] fqfnParts = fqfn.split("/");
+                if (fqfnParts.length != 3) {
+                    throw new RuntimeException(
+                            "Fully qualified function names (FQFNs) must be of the form tenant/namespace/name");
+                }
+                tenant = fqfnParts[0];
+                namespace = fqfnParts[1];
+                functionName = fqfnParts[2];
+            } else {
+                if (null == tenant || null == namespace || null == functionName) {
+                    throw new RuntimeException(
+                            "You must specify a tenant, namespace, and name for the function or a Fully Qualified Function Name (FQFN)");
+                }
+            }
+        }
     }
 
     /**
@@ -135,7 +171,7 @@ abstract class FunctionConfigCommand extends BaseCommand {
         protected String namespace;
         @Parameter(names = "--name", description = "The function's name")
         protected String functionName;
-        @Parameter(names = "--className", description = "The function's class name", required = true)
+        @Parameter(names = "--className", description = "The function's class name")
         protected String className;
         @Parameter(
                 names = "--jar",
@@ -173,6 +209,8 @@ abstract class FunctionConfigCommand extends BaseCommand {
 
         @Override
         void processArguments() throws Exception {
+            super.processArguments();
+
             FunctionConfig.Builder functionConfigBuilder;
 
             // Initialize config builder either from a supplied YAML config file or from scratch
@@ -197,10 +235,7 @@ void processArguments() throws Exception {
             }
 
             if (null != inputs) {
-                String[] topicNames = inputs.split(",");
-                for (int i = 0; i < topicNames.length; ++i) {
-                    functionConfigBuilder.addInputs(topicNames[i]);
-                }
+                Arrays.asList(inputs.split(",")).forEach(functionConfigBuilder::addInputs);
             }
             if (null != customSerdeInputString) {
                 Type type = new TypeToken<Map<String, String>>(){}.getType();
@@ -385,17 +420,6 @@ private void doJavaSubmitChecks(FunctionConfig.Builder functionConfigBuilder) {
             }
         }
 
-        private void parseFullyQualifiedFunctionName(String fqfn, FunctionConfig.Builder functionConfigBuilder) {
-            String[] args = fqfn.split("/");
-            if (args.length != 3) {
-                throw new RuntimeException("Fully qualified function names must be of the form tenant/namespace/name");
-            } else {
-                functionConfigBuilder.setTenant(args[0]);
-                functionConfigBuilder.setNamespace(args[1]);
-                functionConfigBuilder.setName(args[2]);
-            }
-        }
-
         private void doPythonSubmitChecks(FunctionConfig.Builder functionConfigBuilder) {
             if (functionConfigBuilder.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
                 throw new RuntimeException("Effectively-once processing guarantees not yet supported in Python");
@@ -746,4 +770,15 @@ private org.apache.pulsar.functions.proto.Function.FunctionConfig convert(Functi
         Utils.mergeJson(FunctionsImpl.printJson(functionConfig), functionConfigBuilder);
         return functionConfigBuilder.build();
     }
+
+    private void parseFullyQualifiedFunctionName(String fqfn, FunctionConfig.Builder functionConfigBuilder) {
+        String[] args = fqfn.split("/");
+        if (args.length != 3) {
+            throw new RuntimeException("Fully qualified function names (FQFNs) must be of the form tenant/namespace/name");
+        } else {
+            functionConfigBuilder.setTenant(args[0]);
+            functionConfigBuilder.setNamespace(args[1]);
+            functionConfigBuilder.setName(args[2]);
+        }
+    }
 }