/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.pulsar.admin.cli;

import static org.apache.commons.lang.StringUtils.isBlank;
import static org.apache.commons.lang.StringUtils.isNotBlank;
import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParameterException;
import com.beust.jcommander.Parameters;
import com.beust.jcommander.converters.StringConverter;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.annotations.VisibleForTesting;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.reflect.TypeToken;
import java.io.File;
import java.lang.reflect.Field;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.admin.cli.utils.CmdUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.common.functions.ConsumerConfig;
import org.apache.pulsar.common.functions.FunctionConfig;
import org.apache.pulsar.common.functions.FunctionState;
import org.apache.pulsar.common.functions.ProducerConfig;
import org.apache.pulsar.common.functions.Resources;
import org.apache.pulsar.common.functions.UpdateOptionsImpl;
import org.apache.pulsar.common.functions.Utils;
import org.apache.pulsar.common.functions.WindowConfig;
import org.apache.pulsar.common.util.ObjectMapperFactory;

@Slf4j
@Parameters(commandDescription = "Interface for managing Pulsar Functions "
        + "(lightweight, Lambda-style compute processes that work with Pulsar)")
public class CmdFunctions extends CmdBase {
    private final LocalRunner localRunner;
    private final CreateFunction creater;
    private final DeleteFunction deleter;
    private final UpdateFunction updater;
    private final GetFunction getter;
    private final GetFunctionStatus functionStatus;
    @Getter
    private final GetFunctionStats functionStats;
    private final RestartFunction restart;
    private final StopFunction stop;
    private final StartFunction start;
    private final ListFunctions lister;
    private final StateGetter stateGetter;
    private final StatePutter statePutter;
    private final TriggerFunction triggerer;
    private final UploadFunction uploader;
    private final DownloadFunction downloader;

    /**
     * Base command.
     */
    @Getter
    abstract class BaseCommand extends CliCommand {
        @Override
        void run() throws Exception {
            try {
                processArguments();
            } catch (Exception e) {
                System.err.println(e.getMessage());
                System.err.println();
                String chosenCommand = jcommander.getParsedCommand();
                getUsageFormatter().usage(chosenCommand);
                return;
            }
            runCmd();
        }

        void processArguments() throws Exception {}

        abstract void runCmd() throws Exception;
    }

    /**
     * Namespace level command.
     */
    @Getter
    abstract class NamespaceCommand extends BaseCommand {
        @Parameter(names = "--tenant", description = "The tenant of a Pulsar Function")
        protected String tenant;

        @Parameter(names = "--namespace", description = "The namespace of a Pulsar Function")
        protected String namespace;

        @Override
        public void processArguments() {
            if (tenant == null) {
                tenant = PUBLIC_TENANT;
            }
            if (namespace == null) {
                namespace = DEFAULT_NAMESPACE;
            }
        }
    }

    /**
     * Function level command.
     */
    @Getter
    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 tenant of a Pulsar Function")
        protected String tenant;

        @Parameter(names = "--namespace", description = "The namespace of a Pulsar Function")
        protected String namespace;

        @Parameter(names = "--name", description = "The name of a Pulsar Function")
        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 (tenant == null) {
                    tenant = PUBLIC_TENANT;
                }
                if (namespace == null) {
                    namespace = DEFAULT_NAMESPACE;
                }
                if (null == functionName) {
                    throw new RuntimeException(
                            "You must specify a name for the function or a Fully Qualified Function Name (FQFN)");
                }
            }
        }
    }

    /**
     * Commands that require a function config.
     */
    @Getter
    abstract class FunctionDetailsCommand extends BaseCommand {
        @Parameter(names = "--fqfn", description = "The Fully Qualified Function Name (FQFN) for the function")
        protected String fqfn;
        @Parameter(names = "--tenant", description = "The tenant of a Pulsar Function")
        protected String tenant;
        @Parameter(names = "--namespace", description = "The namespace of a Pulsar Function")
        protected String namespace;
        @Parameter(names = "--name", description = "The name of a Pulsar Function")
        protected String functionName;
        // for backwards compatibility purposes
        @Parameter(names = "--className", description = "The class name of a Pulsar Function", hidden = true)
        protected String deprecatedClassName;
        @Parameter(names = "--classname", description = "The class name of a Pulsar Function")
        protected String className;
        @Parameter(names = "--jar", description = "Path to the JAR file for the function "
                + "(if the function is written in Java). It also supports URL path [http/https/file "
                + "(file protocol assumes that file already exists on worker host)/function "
                + "(package URL from packages management service)] from which worker can download the package.",
                listConverter = StringConverter.class)
        protected String jarFile;
        @Parameter(names = "--py", description = "Path to the main Python file/Python Wheel file for the function "
                + "(if the function is written in Python). It also supports URL path [http/https/file "
                + "(file protocol assumes that file already exists on worker host)/function "
                + "(package URL from packages management service)] from which worker can download the package.",
                listConverter = StringConverter.class)
        protected String pyFile;
        @Parameter(names = "--go", description = "Path to the main Go executable binary for the function "
                + "(if the function is written in Go). It also supports URL path [http/https/file "
                + "(file protocol assumes that file already exists on worker host)/function "
                + "(package URL from packages management service)] from which worker can download the package.")
        protected String goFile;
        @Parameter(names = {"-i", "--inputs"}, description = "The input topic or "
                + "topics (multiple topics can be specified as a comma-separated list) of a Pulsar Function")
        protected String inputs;
        // for backwards compatibility purposes
        @Parameter(names = "--topicsPattern", description = "TopicsPattern to consume from list of topics "
                + "under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. "
                + "Add SerDe class name for a pattern in --custom-serde-inputs (supported for java fun only)",
                hidden = true)
        protected String deprecatedTopicsPattern;
        @Parameter(names = "--topics-pattern", description = "The topic pattern to consume from list of topics "
                + "under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. "
                + "Add SerDe class name for a pattern in --custom-serde-inputs (supported for java fun only)")
        protected String topicsPattern;

        @Parameter(names = {"-o", "--output"},
                description = "The output topic of a Pulsar Function (If none is specified, no output is written)")
        protected String output;
        @Parameter(names = "--producer-config", description = "The custom producer configuration (as a JSON string)")
        protected String producerConfig;
        // for backwards compatibility purposes
        @Parameter(names = "--logTopic",
                description = "The topic to which the logs of a Pulsar Function are produced", hidden = true)
        protected String deprecatedLogTopic;
        @Parameter(names = "--log-topic", description = "The topic to which the logs of a Pulsar Function are produced")
        protected String logTopic;

        @Parameter(names = {"-st", "--schema-type"}, description = "The builtin schema type or "
                + "custom schema class name to be used for messages output by the function")
        protected String schemaType = "";

        // for backwards compatibility purposes
        @Parameter(names = "--customSerdeInputs",
                description = "The map of input topics to SerDe class names (as a JSON string)", hidden = true)
        protected String deprecatedCustomSerdeInputString;
        @Parameter(names = "--custom-serde-inputs",
                description = "The map of input topics to SerDe class names (as a JSON string)")
        protected String customSerdeInputString;
        @Parameter(names = "--custom-schema-inputs",
                description = "The map of input topics to Schema properties (as a JSON string)")
        protected String customSchemaInputString;
        @Parameter(names = "--custom-schema-outputs",
                description = "The map of input topics to Schema properties (as a JSON string)")
        protected String customSchemaOutputString;
        @Parameter(names = "--input-specs",
                description = "The map of inputs to custom configuration (as a JSON string)")
        protected String inputSpecs;
        // for backwards compatibility purposes
        @Parameter(names = "--outputSerdeClassName",
                description = "The SerDe class to be used for messages output by the function", hidden = true)
        protected String deprecatedOutputSerdeClassName;
        @Parameter(names = "--output-serde-classname",
                description = "The SerDe class to be used for messages output by the function")
        protected String outputSerdeClassName;
        // for backwards compatibility purposes
        @Parameter(names = "--functionConfigFile", description = "The path to a YAML config file that specifies "
                + "the configuration of a Pulsar Function", hidden = true)
        protected String deprecatedFnConfigFile;
        @Parameter(names = "--function-config-file",
                description = "The path to a YAML config file that specifies the configuration of a Pulsar Function")
        protected String fnConfigFile;
        // for backwards compatibility purposes
        @Parameter(names = "--processingGuarantees", description = "The processing guarantees (aka delivery semantics) "
                + "applied to the function", hidden = true)
        protected FunctionConfig.ProcessingGuarantees deprecatedProcessingGuarantees;
        @Parameter(names = "--processing-guarantees",
                description = "The processing guarantees (aka delivery semantics) applied to the function")
        protected FunctionConfig.ProcessingGuarantees processingGuarantees;
        // for backwards compatibility purposes
        @Parameter(names = "--userConfig", description = "User-defined config key/values", hidden = true)
        protected String deprecatedUserConfigString;
        @Parameter(names = "--user-config", description = "User-defined config key/values")
        protected String userConfigString;
        @Parameter(names = "--retainOrdering",
                description = "Function consumes and processes messages in order", hidden = true)
        protected Boolean deprecatedRetainOrdering;
        @Parameter(names = "--retain-ordering", description = "Function consumes and processes messages in order")
        protected Boolean retainOrdering;
        @Parameter(names = "--retain-key-ordering",
                description = "Function consumes and processes messages in key order")
        protected Boolean retainKeyOrdering;
        @Parameter(names = "--batch-builder", description = "BatcherBuilder provides two types of "
                + "batch construction methods, DEFAULT and KEY_BASED. The default value is: DEFAULT")
        protected String batchBuilder;
        @Parameter(names = "--forward-source-message-property", description = "Forwarding input message's properties "
                + "to output topic when processing (use false to disable it)", arity = 1)
        protected Boolean forwardSourceMessageProperty = true;
        @Parameter(names = "--subs-name", description = "Pulsar source subscription name if user wants a specific "
                + "subscription-name for input-topic consumer")
        protected String subsName;
        @Parameter(names = "--subs-position", description = "Pulsar source subscription position if user wants to "
                + "consume messages from the specified location")
        protected SubscriptionInitialPosition subsPosition;
        @Parameter(names = "--parallelism", description = "The parallelism factor of a Pulsar Function "
                + "(i.e. the number of function instances to run)")
        protected Integer parallelism;
        @Parameter(names = "--cpu", description = "The cpu in cores that need to be allocated "
                + "per function instance(applicable only to docker runtime)")
        protected Double cpu;
        @Parameter(names = "--ram", description = "The ram in bytes that need to be allocated "
                + "per function instance(applicable only to process/docker runtime)")
        protected Long ram;
        @Parameter(names = "--disk", description = "The disk in bytes that need to be allocated "
                + "per function instance(applicable only to docker runtime)")
        protected Long disk;
        // for backwards compatibility purposes
        @Parameter(names = "--windowLengthCount", description = "The number of messages per window", hidden = true)
        protected Integer deprecatedWindowLengthCount;
        @Parameter(names = "--window-length-count", description = "The number of messages per window")
        protected Integer windowLengthCount;
        // for backwards compatibility purposes
        @Parameter(names = "--windowLengthDurationMs",
                description = "The time duration of the window in milliseconds", hidden = true)
        protected Long deprecatedWindowLengthDurationMs;
        @Parameter(names = "--window-length-duration-ms",
                description = "The time duration of the window in milliseconds")
        protected Long windowLengthDurationMs;
        // for backwards compatibility purposes
        @Parameter(names = "--slidingIntervalCount",
                description = "The number of messages after which the window slides", hidden = true)
        protected Integer deprecatedSlidingIntervalCount;
        @Parameter(names = "--sliding-interval-count",
                description = "The number of messages after which the window slides")
        protected Integer slidingIntervalCount;
        // for backwards compatibility purposes
        @Parameter(names = "--slidingIntervalDurationMs",
                description = "The time duration after which the window slides", hidden = true)
        protected Long deprecatedSlidingIntervalDurationMs;
        @Parameter(names = "--sliding-interval-duration-ms",
                description = "The time duration after which the window slides")
        protected Long slidingIntervalDurationMs;
        // for backwards compatibility purposes
        @Parameter(names = "--autoAck",
                description = "Whether or not the framework acknowledges messages automatically", hidden = true)
        protected Boolean deprecatedAutoAck = null;
        @Parameter(names = "--auto-ack",
                description = "Whether or not the framework acknowledges messages automatically", arity = 1)
        protected Boolean autoAck;
        // for backwards compatibility purposes
        @Parameter(names = "--timeoutMs", description = "The message timeout in milliseconds", hidden = true)
        protected Long deprecatedTimeoutMs;
        @Parameter(names = "--timeout-ms", description = "The message timeout in milliseconds")
        protected Long timeoutMs;
        @Parameter(names = "--max-message-retries",
                description = "How many times should we try to process a message before giving up")
        protected Integer maxMessageRetries;
        @Parameter(names = "--custom-runtime-options", description = "A string that encodes options to "
                + "customize the runtime, see docs for configured runtime for details")
        protected String customRuntimeOptions;
        @Parameter(names = "--secrets", description = "The map of secretName to an object that encapsulates "
                + "how the secret is fetched by the underlying secrets provider")
        protected String secretsString;
        @Parameter(names = "--dead-letter-topic",
                description = "The topic where messages that are not processed successfully are sent to")
        protected String deadLetterTopic;
        protected FunctionConfig functionConfig;
        protected String userCodeFile;

        private void mergeArgs() {
            if (isBlank(className) && !isBlank(deprecatedClassName)) {
                className = deprecatedClassName;
            }
            if (isBlank(topicsPattern) && !isBlank(deprecatedTopicsPattern)) {
                topicsPattern = deprecatedTopicsPattern;
            }
            if (isBlank(logTopic) && !isBlank(deprecatedLogTopic)) {
                logTopic = deprecatedLogTopic;
            }
            if (isBlank(outputSerdeClassName) && !isBlank(deprecatedOutputSerdeClassName)) {
                outputSerdeClassName = deprecatedOutputSerdeClassName;
            }
            if (isBlank(customSerdeInputString) && !isBlank(deprecatedCustomSerdeInputString)) {
                customSerdeInputString = deprecatedCustomSerdeInputString;
            }

            if (isBlank(fnConfigFile) && !isBlank(deprecatedFnConfigFile)) {
                fnConfigFile = deprecatedFnConfigFile;
            }
            if (processingGuarantees == null && deprecatedProcessingGuarantees != null) {
                processingGuarantees = deprecatedProcessingGuarantees;
            }
            if (isBlank(userConfigString) && !isBlank(deprecatedUserConfigString)) {
                userConfigString = deprecatedUserConfigString;
            }
            if (retainOrdering == null && deprecatedRetainOrdering != null) {
                retainOrdering = deprecatedRetainOrdering;
            }
            if (windowLengthCount == null && deprecatedWindowLengthCount != null) {
                windowLengthCount = deprecatedWindowLengthCount;
            }
            if (windowLengthDurationMs == null && deprecatedWindowLengthDurationMs != null) {
                windowLengthDurationMs = deprecatedWindowLengthDurationMs;
            }
            if (slidingIntervalCount == null && deprecatedSlidingIntervalCount != null) {
                slidingIntervalCount = deprecatedSlidingIntervalCount;
            }
            if (slidingIntervalDurationMs == null && deprecatedSlidingIntervalDurationMs != null) {
                slidingIntervalDurationMs = deprecatedSlidingIntervalDurationMs;
            }
            if (autoAck == null && deprecatedAutoAck != null) {
                autoAck = deprecatedAutoAck;
            }
            if (timeoutMs == null && deprecatedTimeoutMs != null) {
                timeoutMs = deprecatedTimeoutMs;
            }
        }

        @Override
        void processArguments() throws Exception {
            super.processArguments();
            // merge deprecated args with new args
            mergeArgs();

            // Initialize config builder either from a supplied YAML config file or from scratch
            if (null != fnConfigFile) {
                functionConfig = CmdUtils.loadConfig(fnConfigFile, FunctionConfig.class);
            } else {
                functionConfig = new FunctionConfig();
            }

            if (null != fqfn) {
                parseFullyQualifiedFunctionName(fqfn, functionConfig);
            } else {
                if (null != tenant) {
                    functionConfig.setTenant(tenant);
                }
                if (null != namespace) {
                    functionConfig.setNamespace(namespace);
                }
                if (null != functionName) {
                    functionConfig.setName(functionName);
                }
            }

            if (null != inputs) {
                List<String> inputTopics = Arrays.asList(inputs.split(","));
                functionConfig.setInputs(inputTopics);
            }
            if (null != customSerdeInputString) {
                Type type = new TypeToken<Map<String, String>>() {}.getType();
                Map<String, String> customSerdeInputMap = new Gson().fromJson(customSerdeInputString, type);
                functionConfig.setCustomSerdeInputs(customSerdeInputMap);
            }
            if (null != customSchemaInputString) {
                Type type = new TypeToken<Map<String, String>>() {}.getType();
                Map<String, String> customschemaInputMap = new Gson().fromJson(customSchemaInputString, type);
                functionConfig.setCustomSchemaInputs(customschemaInputMap);
            }
            if (null != customSchemaOutputString) {
                Type type = new TypeToken<Map<String, String>>() {}.getType();
                Map<String, String> customSchemaOutputMap = new Gson().fromJson(customSchemaOutputString, type);
                functionConfig.setCustomSchemaOutputs(customSchemaOutputMap);
            }
            if (null != inputSpecs) {
                Type type = new TypeToken<Map<String, ConsumerConfig>>() {}.getType();
                functionConfig.setInputSpecs(new Gson().fromJson(inputSpecs, type));
            }
            if (null != topicsPattern) {
                functionConfig.setTopicsPattern(topicsPattern);
            }
            if (null != output) {
                functionConfig.setOutput(output);
            }
            if (null != producerConfig) {
                Type type = new TypeToken<ProducerConfig>() {}.getType();
                functionConfig.setProducerConfig(new Gson().fromJson(producerConfig, type));
            }
            if (null != logTopic) {
                functionConfig.setLogTopic(logTopic);
            }
            if (null != className) {
                functionConfig.setClassName(className);
            }
            if (null != outputSerdeClassName) {
                functionConfig.setOutputSerdeClassName(outputSerdeClassName);
            }

            if (null != schemaType) {
                functionConfig.setOutputSchemaType(schemaType);
            }
            if (null != processingGuarantees) {
                functionConfig.setProcessingGuarantees(processingGuarantees);
            }

            if (null != retainOrdering) {
                functionConfig.setRetainOrdering(retainOrdering);
            }

            if (null != retainKeyOrdering) {
                functionConfig.setRetainKeyOrdering(retainKeyOrdering);
            }

            if (isNotBlank(batchBuilder)) {
                functionConfig.setBatchBuilder(batchBuilder);
            }

            if (null != forwardSourceMessageProperty) {
                functionConfig.setForwardSourceMessageProperty(forwardSourceMessageProperty);
            }

            if (isNotBlank(subsName)) {
                functionConfig.setSubName(subsName);
            }

            if (null != subsPosition) {
                functionConfig.setSubscriptionPosition(subsPosition);
            }

            if (null != userConfigString) {
                Type type = new TypeToken<Map<String, String>>() {}.getType();
                Map<String, Object> userConfigMap = new Gson().fromJson(userConfigString, type);
                if (userConfigMap == null) {
                    userConfigMap = new HashMap<>();
                }
                functionConfig.setUserConfig(userConfigMap);
            }

            if (parallelism != null) {
                functionConfig.setParallelism(parallelism);
            }

            Resources resources = functionConfig.getResources();
            if (cpu != null) {
                if (resources == null) {
                    resources = new Resources();
                }
                resources.setCpu(cpu);
            }

            if (ram != null) {
                if (resources == null) {
                    resources = new Resources();
                }
                resources.setRam(ram);
            }

            if (disk != null) {
                if (resources == null) {
                    resources = new Resources();
                }
                resources.setDisk(disk);
            }
            if (resources != null) {
                functionConfig.setResources(resources);
            }

            if (timeoutMs != null) {
                functionConfig.setTimeoutMs(timeoutMs);
            }

            if (customRuntimeOptions != null) {
                functionConfig.setCustomRuntimeOptions(customRuntimeOptions);
            }

            if (secretsString != null) {
                Type type = new TypeToken<Map<String, Object>>() {}.getType();
                Map<String, Object> secretsMap = new Gson().fromJson(secretsString, type);
                if (secretsMap == null) {
                    secretsMap = Collections.emptyMap();
                }
                functionConfig.setSecrets(secretsMap);
            }

            // window configs
            WindowConfig windowConfig = functionConfig.getWindowConfig();
            if (null != windowLengthCount) {
                if (windowConfig == null) {
                    windowConfig = new WindowConfig();
                }
                windowConfig.setWindowLengthCount(windowLengthCount);
            }
            if (null != windowLengthDurationMs) {
                if (windowConfig == null) {
                    windowConfig = new WindowConfig();
                }
                windowConfig.setWindowLengthDurationMs(windowLengthDurationMs);
            }
            if (null != slidingIntervalCount) {
                if (windowConfig == null) {
                    windowConfig = new WindowConfig();
                }
                windowConfig.setSlidingIntervalCount(slidingIntervalCount);
            }
            if (null != slidingIntervalDurationMs) {
                if (windowConfig == null) {
                    windowConfig = new WindowConfig();
                }
                windowConfig.setSlidingIntervalDurationMs(slidingIntervalDurationMs);
            }

            functionConfig.setWindowConfig(windowConfig);

            if (autoAck != null) {
                functionConfig.setAutoAck(autoAck);
            }

            if (null != maxMessageRetries) {
                functionConfig.setMaxMessageRetries(maxMessageRetries);
            }
            if (null != deadLetterTopic) {
                functionConfig.setDeadLetterTopic(deadLetterTopic);
            }

            if (null != jarFile) {
                functionConfig.setJar(jarFile);
            }

            if (null != pyFile) {
                functionConfig.setPy(pyFile);
            }

            if (null != goFile) {
                functionConfig.setGo(goFile);
            }

            if (functionConfig.getJar() != null) {
                userCodeFile = functionConfig.getJar();
            } else if (functionConfig.getPy() != null) {
                userCodeFile = functionConfig.getPy();
            } else if (functionConfig.getGo() != null) {
                userCodeFile = functionConfig.getGo();
            }

            // check if configs are valid
            validateFunctionConfigs(functionConfig);
        }

        protected void validateFunctionConfigs(FunctionConfig functionConfig) {
            // go doesn't need className
            if (functionConfig.getRuntime() == FunctionConfig.Runtime.PYTHON
                    || functionConfig.getRuntime() == FunctionConfig.Runtime.JAVA){
                if (StringUtils.isEmpty(functionConfig.getClassName())) {
                    throw new IllegalArgumentException("No Function Classname specified");
                }
            }
            if (StringUtils.isEmpty(functionConfig.getName())) {
                org.apache.pulsar.common.functions.Utils.inferMissingFunctionName(functionConfig);
            }
            if (StringUtils.isEmpty(functionConfig.getTenant())) {
                org.apache.pulsar.common.functions.Utils.inferMissingTenant(functionConfig);
            }
            if (StringUtils.isEmpty(functionConfig.getNamespace())) {
                org.apache.pulsar.common.functions.Utils.inferMissingNamespace(functionConfig);
            }

            if (isNotBlank(functionConfig.getJar()) && isNotBlank(functionConfig.getPy())
                    && isNotBlank(functionConfig.getGo())) {
                throw new ParameterException("Either a Java jar or a Python file or a Go executable binary needs to"
                        + " be specified for the function. Cannot specify both.");
            }

            if (isBlank(functionConfig.getJar()) && isBlank(functionConfig.getPy())
                    && isBlank(functionConfig.getGo())) {
                throw new ParameterException("Either a Java jar or a Python file or a Go executable binary needs to"
                        + " be specified for the function. Please specify one.");
            }

            if (!isBlank(functionConfig.getJar()) && !Utils.isFunctionPackageUrlSupported(functionConfig.getJar())
                    && !new File(functionConfig.getJar()).exists()) {
                throw new ParameterException("The specified jar file does not exist");
            }
            if (!isBlank(functionConfig.getPy()) && !Utils.isFunctionPackageUrlSupported(functionConfig.getPy())
                    && !new File(functionConfig.getPy()).exists()) {
                throw new ParameterException("The specified python file does not exist");
            }
            if (!isBlank(functionConfig.getGo()) && !Utils.isFunctionPackageUrlSupported(functionConfig.getGo())
                    && !new File(functionConfig.getGo()).exists()) {
                throw new ParameterException("The specified go executable binary does not exist");
            }
        }
    }

    @Parameters(commandDescription = "Run a Pulsar Function locally, rather than deploy to a Pulsar cluster)")
    class LocalRunner extends FunctionDetailsCommand {

        // TODO: this should become BookKeeper URL and it should be fetched from Pulsar client.
        // for backwards compatibility purposes
        @Parameter(names = "--stateStorageServiceUrl", description = "The URL for the state storage service "
                + "(the default is Apache BookKeeper)", hidden = true)
        protected String deprecatedStateStorageServiceUrl;
        @Parameter(names = "--state-storage-service-url", description = "The URL for the state storage service "
                + "(the default is Apache BookKeeper)")
        protected String stateStorageServiceUrl;
        // for backwards compatibility purposes
        @Parameter(names = "--brokerServiceUrl", description = "The URL for Pulsar broker", hidden = true)
        protected String deprecatedBrokerServiceUrl;
        @Parameter(names = "--broker-service-url", description = "The URL for Pulsar broker")
        protected String brokerServiceUrl;
        @Parameter(names = "--web-service-url", description = "The URL for Pulsar web service")
        protected String webServiceUrl = null;
        // for backwards compatibility purposes
        @Parameter(names = "--clientAuthPlugin", description = "Client authentication plugin using "
                + "which function-process can connect to broker", hidden = true)
        protected String deprecatedClientAuthPlugin;
        @Parameter(names = "--client-auth-plugin",
                description = "Client authentication plugin using which function-process can connect to broker")
        protected String clientAuthPlugin;
        // for backwards compatibility purposes
        @Parameter(names = "--clientAuthParams", description = "Client authentication param", hidden = true)
        protected String deprecatedClientAuthParams;
        @Parameter(names = "--client-auth-params", description = "Client authentication param")
        protected String clientAuthParams;
        // for backwards compatibility purposes
        @Parameter(names = "--use_tls", description = "Use tls connection", hidden = true)
        protected Boolean deprecatedUseTls = null;
        @Parameter(names = "--use-tls", description = "Use tls connection")
        protected boolean useTls;
        // for backwards compatibility purposes
        @Parameter(names = "--tls_allow_insecure", description = "Allow insecure tls connection", hidden = true)
        protected Boolean deprecatedTlsAllowInsecureConnection = null;
        @Parameter(names = "--tls-allow-insecure", description = "Allow insecure tls connection")
        protected boolean tlsAllowInsecureConnection;
        // for backwards compatibility purposes
        @Parameter(names = "--hostname_verification_enabled",
                description = "Enable hostname verification", hidden = true)
        protected Boolean deprecatedTlsHostNameVerificationEnabled = null;
        @Parameter(names = "--hostname-verification-enabled", description = "Enable hostname verification")
        protected boolean tlsHostNameVerificationEnabled;
        // for backwards compatibility purposes
        @Parameter(names = "--tls_trust_cert_path", description = "tls trust cert file path", hidden = true)
        protected String deprecatedTlsTrustCertFilePath;
        @Parameter(names = "--tls-trust-cert-path", description = "tls trust cert file path")
        protected String tlsTrustCertFilePath;
        // for backwards compatibility purposes
        @Parameter(names = "--instanceIdOffset", description = "Start the instanceIds from this offset", hidden = true)
        protected Integer deprecatedInstanceIdOffset = null;
        @Parameter(names = "--instance-id-offset", description = "Start the instanceIds from this offset")
        protected Integer instanceIdOffset = 0;
        @Parameter(names = "--runtime", description = "either THREAD or PROCESS. Only applies for Java functions")
        protected String runtime;
        @Parameter(names = "--secrets-provider-classname", description = "Whats the classname for secrets provider")
        protected String secretsProviderClassName;
        @Parameter(names = "--secrets-provider-config",
                description = "Config that needs to be passed to secrets provider")
        protected String secretsProviderConfig;
        @Parameter(names = "--metrics-port-start", description = "The starting port range for metrics server")
        protected String metricsPortStart;

        private void mergeArgs() {
            if (isBlank(stateStorageServiceUrl) && !isBlank(deprecatedStateStorageServiceUrl)) {
                stateStorageServiceUrl = deprecatedStateStorageServiceUrl;
            }
            if (isBlank(brokerServiceUrl) && !isBlank(deprecatedBrokerServiceUrl)) {
                brokerServiceUrl = deprecatedBrokerServiceUrl;
            }
            if (isBlank(clientAuthPlugin) && !isBlank(deprecatedClientAuthPlugin)) {
                clientAuthPlugin = deprecatedClientAuthPlugin;
            }
            if (isBlank(clientAuthParams) && !isBlank(deprecatedClientAuthParams)) {
                clientAuthParams = deprecatedClientAuthParams;
            }
            if (!useTls && deprecatedUseTls != null) {
                useTls = deprecatedUseTls;
            }
            if (!tlsAllowInsecureConnection && deprecatedTlsAllowInsecureConnection != null) {
                tlsAllowInsecureConnection = deprecatedTlsAllowInsecureConnection;
            }
            if (!tlsHostNameVerificationEnabled && deprecatedTlsHostNameVerificationEnabled != null) {
                tlsHostNameVerificationEnabled = deprecatedTlsHostNameVerificationEnabled;
            }
            if (isBlank(tlsTrustCertFilePath) && !isBlank(deprecatedTlsTrustCertFilePath)) {
                tlsTrustCertFilePath = deprecatedTlsTrustCertFilePath;
            }
            if (instanceIdOffset == null && deprecatedInstanceIdOffset != null) {
                instanceIdOffset = deprecatedInstanceIdOffset;
            }
        }

        @Override
        void runCmd() throws Exception {
            // merge deprecated args with new args
            mergeArgs();
            List<String> localRunArgs = new LinkedList<>();
            localRunArgs.add(System.getenv("PULSAR_HOME") + "/bin/function-localrunner");
            localRunArgs.add("--functionConfig");
            localRunArgs.add(new Gson().toJson(functionConfig));
            for (Field field : this.getClass().getDeclaredFields()) {
                if (field.getName().startsWith("DEPRECATED")) {
                    continue;
                }
                if (field.getName().contains("$")) {
                    continue;
                }
                Object value = field.get(this);
                if (value != null) {
                    localRunArgs.add("--" + field.getName());
                    localRunArgs.add(value.toString());
                }
            }
            ProcessBuilder processBuilder = new ProcessBuilder(localRunArgs).inheritIO();
            Process process = processBuilder.start();
            process.waitFor();
        }
    }

    @Parameters(commandDescription = "Create a Pulsar Function in cluster mode (deploy it on a Pulsar cluster)")
    class CreateFunction extends FunctionDetailsCommand {
        @Override
        void runCmd() throws Exception {
            if (Utils.isFunctionPackageUrlSupported(functionConfig.getJar())) {
                getAdmin().functions().createFunctionWithUrl(functionConfig, functionConfig.getJar());
            } else if (Utils.isFunctionPackageUrlSupported(functionConfig.getPy())) {
                getAdmin().functions().createFunctionWithUrl(functionConfig, functionConfig.getPy());
            } else if (Utils.isFunctionPackageUrlSupported(functionConfig.getGo())) {
                getAdmin().functions().createFunctionWithUrl(functionConfig, functionConfig.getGo());
            } else {
                getAdmin().functions().createFunction(functionConfig, userCodeFile);
            }

            print("Created successfully");
        }
    }

    @Parameters(commandDescription = "Fetch information about a Pulsar Function")
    class GetFunction extends FunctionCommand {
        @Override
        void runCmd() throws Exception {
            FunctionConfig functionConfig = getAdmin().functions().getFunction(tenant, namespace, functionName);
            Gson gson = new GsonBuilder().setPrettyPrinting().create();
            System.out.println(gson.toJson(functionConfig));
        }
    }

    @Parameters(commandDescription = "Check the current status of a Pulsar Function")
    class GetFunctionStatus extends FunctionCommand {

        @Parameter(names = "--instance-id", description = "The function instanceId "
                + "(Get-status of all instances if instance-id is not provided)")
        protected String instanceId;

        @Override
        void runCmd() throws Exception {
            if (isBlank(instanceId)) {
                print(getAdmin().functions().getFunctionStatus(tenant, namespace, functionName));
            } else {
                print(getAdmin().functions()
                        .getFunctionStatus(tenant, namespace, functionName, Integer.parseInt(instanceId)));
            }
        }
    }

    @Parameters(commandDescription = "Get the current stats of a Pulsar Function")
    class GetFunctionStats extends FunctionCommand {

        @Parameter(names = "--instance-id", description = "The function instanceId "
                + "(Get-stats of all instances if instance-id is not provided)")
        protected String instanceId;

        @Override
        void runCmd() throws Exception {

            if (isBlank(instanceId)) {
                print(getAdmin().functions().getFunctionStats(tenant, namespace, functionName));
            } else {
               print(getAdmin().functions()
                       .getFunctionStats(tenant, namespace, functionName, Integer.parseInt(instanceId)));
            }
        }
    }

    @Parameters(commandDescription = "Restart function instance")
    class RestartFunction extends FunctionCommand {

        @Parameter(names = "--instance-id", description = "The function instanceId "
                + "(restart all instances if instance-id is not provided)")
        protected String instanceId;

        @Override
        void runCmd() throws Exception {
            if (isNotBlank(instanceId)) {
                try {
                    getAdmin().functions()
                            .restartFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
                } catch (NumberFormatException e) {
                    System.err.println("instance-id must be a number");
                }
            } else {
                getAdmin().functions().restartFunction(tenant, namespace, functionName);
            }
            System.out.println("Restarted successfully");
        }
    }

    @Parameters(commandDescription = "Stops function instance")
    class StopFunction extends FunctionCommand {

        @Parameter(names = "--instance-id", description = "The function instanceId "
                + "(stop all instances if instance-id is not provided)")
        protected String instanceId;

        @Override
        void runCmd() throws Exception {
            if (isNotBlank(instanceId)) {
                try {
                    getAdmin().functions().stopFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
                } catch (NumberFormatException e) {
                    System.err.println("instance-id must be a number");
                }
            } else {
                getAdmin().functions().stopFunction(tenant, namespace, functionName);
            }
            System.out.println("Stopped successfully");
        }
    }

    @Parameters(commandDescription = "Starts a stopped function instance")
    class StartFunction extends FunctionCommand {

        @Parameter(names = "--instance-id", description = "The function instanceId "
                + "(start all instances if instance-id is not provided)")
        protected String instanceId;

        @Override
        void runCmd() throws Exception {
            if (isNotBlank(instanceId)) {
                try {
                    getAdmin().functions().startFunction(tenant, namespace, functionName, Integer.parseInt(instanceId));
                } catch (NumberFormatException e) {
                    System.err.println("instance-id must be a number");
                }
            } else {
                getAdmin().functions().startFunction(tenant, namespace, functionName);
            }
            System.out.println("Started successfully");
        }
    }

    @Parameters(commandDescription = "Delete a Pulsar Function that is running on a Pulsar cluster")
    class DeleteFunction extends FunctionCommand {
        @Override
        void runCmd() throws Exception {
            getAdmin().functions().deleteFunction(tenant, namespace, functionName);
            print("Deleted successfully");
        }
    }

    @Parameters(commandDescription = "Update a Pulsar Function that has been deployed to a Pulsar cluster")
    class UpdateFunction extends FunctionDetailsCommand {

        @Parameter(names = "--update-auth-data", description = "Whether or not to update the auth data")
        protected boolean updateAuthData;

        @Override
        protected void validateFunctionConfigs(FunctionConfig functionConfig) {
            if (StringUtils.isEmpty(functionConfig.getClassName())) {
                if (StringUtils.isEmpty(functionConfig.getName())) {
                    throw new IllegalArgumentException("Function Name not provided");
                }
            } else if (StringUtils.isEmpty(functionConfig.getName())) {
                org.apache.pulsar.common.functions.Utils.inferMissingFunctionName(functionConfig);
            }
            if (StringUtils.isEmpty(functionConfig.getTenant())) {
                org.apache.pulsar.common.functions.Utils.inferMissingTenant(functionConfig);
            }
            if (StringUtils.isEmpty(functionConfig.getNamespace())) {
                org.apache.pulsar.common.functions.Utils.inferMissingNamespace(functionConfig);
            }
        }

        @Override
        void runCmd() throws Exception {

            UpdateOptionsImpl updateOptions = new UpdateOptionsImpl();
            updateOptions.setUpdateAuthData(updateAuthData);
            if (Utils.isFunctionPackageUrlSupported(functionConfig.getJar())) {
                getAdmin().functions().updateFunctionWithUrl(functionConfig, functionConfig.getJar(), updateOptions);
            } else {
                getAdmin().functions().updateFunction(functionConfig, userCodeFile, updateOptions);
            }
            print("Updated successfully");
        }
    }

    @Parameters(commandDescription = "List all Pulsar Functions running under a specific tenant and namespace")
    class ListFunctions extends NamespaceCommand {
        @Override
        void runCmd() throws Exception {
            print(getAdmin().functions().getFunctions(tenant, namespace));
        }
    }

    @Parameters(commandDescription = "Fetch the current state associated with a Pulsar Function")
    class StateGetter extends FunctionCommand {

        @Parameter(names = { "-k", "--key" }, description = "Key name of State")
        private String key = null;

        @Parameter(names = { "-w", "--watch" }, description = "Watch for changes in the value associated with a key "
                + "for a Pulsar Function")
        private boolean watch = false;

        @Override
        void runCmd() throws Exception {
            if (isBlank(key)) {
                throw new ParameterException("State key needs to be specified");
            }
            do {
                try {
                    FunctionState functionState = getAdmin().functions()
                                                       .getFunctionState(tenant, namespace, functionName, key);
                    Gson gson = new GsonBuilder().setPrettyPrinting().create();
                    System.out.println(gson.toJson(functionState));
                } catch (PulsarAdminException pae) {
                    if (pae.getStatusCode() == 404 && watch) {
                        System.err.println(pae.getMessage());
                    } else {
                        throw pae;
                    }
                }
                if (watch) {
                    Thread.sleep(1000);
                }
            } while (watch);
        }
    }

    @Parameters(commandDescription = "Put the state associated with a Pulsar Function")
    class StatePutter extends FunctionCommand {

        @Parameter(names = { "-s", "--state" }, description = "The FunctionState that needs to be put", required = true)
        private String state = null;

        @Override
        void runCmd() throws Exception {
            TypeReference<FunctionState> typeRef = new TypeReference<FunctionState>() {};
            FunctionState stateRepr = ObjectMapperFactory.getThreadLocal().readValue(state, typeRef);
            getAdmin().functions()
                    .putFunctionState(tenant, namespace, functionName, stateRepr);
        }
    }

    @Parameters(commandDescription = "Trigger the specified Pulsar Function with a supplied value")
    class TriggerFunction extends FunctionCommand {
        // for backward compatibility purposes
        @Parameter(names = "--triggerValue",
                description = "The value with which you want to trigger the function", hidden = true)
        protected String deprecatedTriggerValue;
        @Parameter(names = "--trigger-value", description = "The value with which you want to trigger the function")
        protected String triggerValue;
        // for backward compatibility purposes
        @Parameter(names = "--triggerFile", description = "The path to the file that contains the data with which "
                + "you want to trigger the function", hidden = true)
        protected String deprecatedTriggerFile;
        @Parameter(names = "--trigger-file", description = "The path to the file that contains the data with which "
                + "you want to trigger the function")
        protected String triggerFile;
        @Parameter(names = "--topic", description = "The specific topic name that the function consumes from that"
                + " you want to inject the data to")
        protected String topic;

        public void mergeArgs() {
            if (isBlank(triggerValue) && !isBlank(deprecatedTriggerValue)) {
                triggerValue = deprecatedTriggerValue;
            }
            if (isBlank(triggerFile) && !isBlank(deprecatedTriggerFile)) {
                triggerFile = deprecatedTriggerFile;
            }
        }

        @Override
        void runCmd() throws Exception {
            // merge deprecated args with new args
            mergeArgs();
            if (triggerFile == null && triggerValue == null) {
                throw new ParameterException("Either a trigger value or a trigger filepath needs to be specified");
            }
            String retval = getAdmin().functions()
                    .triggerFunction(tenant, namespace, functionName, topic, triggerValue, triggerFile);
            System.out.println(retval);
        }
    }

    @Parameters(commandDescription = "Upload File Data to Pulsar", hidden = true)
    class UploadFunction extends BaseCommand {
        // for backward compatibility purposes
        @Parameter(
                names = "--sourceFile",
                description = "The file whose contents need to be uploaded",
                listConverter = StringConverter.class, hidden = true)
        protected String deprecatedSourceFile;
        @Parameter(
                names = "--source-file",
                description = "The file whose contents need to be uploaded",
                listConverter = StringConverter.class)
        protected String sourceFile;
        @Parameter(
                names = "--path",
                description = "Path where the contents need to be stored",
                listConverter = StringConverter.class, required = true)
        protected String path;

        private void mergeArgs() {
            if (isBlank(sourceFile) && !isBlank(deprecatedSourceFile)) {
                sourceFile = deprecatedSourceFile;
            }
        }

        @Override
        void runCmd() throws Exception {
            // merge deprecated args with new args
            mergeArgs();
            if (StringUtils.isBlank(sourceFile)) {
                throw new ParameterException("--source-file needs to be specified");
            }
            getAdmin().functions().uploadFunction(sourceFile, path);
            print("Uploaded successfully");
        }
    }

    @Parameters(commandDescription = "Download File Data from Pulsar", hidden = true)
    class DownloadFunction extends FunctionCommand {
        // for backward compatibility purposes
        @Parameter(
                names = "--destinationFile",
                description = "The file to store downloaded content",
                listConverter = StringConverter.class, hidden = true)
        protected String deprecatedDestinationFile;
        @Parameter(
                names = "--destination-file",
                description = "The file to store downloaded content",
                listConverter = StringConverter.class)
        protected String destinationFile;
        @Parameter(
                names = "--path",
                description = "Path to store the content",
                listConverter = StringConverter.class, required = false, hidden = true)
        protected String path;

        private void mergeArgs() {
            if (isBlank(destinationFile) && !isBlank(deprecatedDestinationFile)) {
                destinationFile = deprecatedDestinationFile;
            }
        }

        @Override
        void processArguments() throws Exception {
            if (path == null) {
                super.processArguments();
            }
        }

        @Override
        void runCmd() throws Exception {
            // merge deprecated args with new args
            mergeArgs();
            if (StringUtils.isBlank(destinationFile)) {
                throw new ParameterException("--destination-file needs to be specified");
            }
            if (path != null) {
                getAdmin().functions().downloadFunction(destinationFile, path);
            } else {
                getAdmin().functions().downloadFunction(destinationFile, tenant, namespace, functionName);
            }
            print("Downloaded successfully");
        }
    }

    public CmdFunctions(Supplier<PulsarAdmin> admin) throws PulsarClientException {
        super("functions", admin);
        localRunner = new LocalRunner();
        creater = new CreateFunction();
        deleter = new DeleteFunction();
        updater = new UpdateFunction();
        getter = new GetFunction();
        functionStatus = new GetFunctionStatus();
        functionStats = new GetFunctionStats();
        lister = new ListFunctions();
        stateGetter = new StateGetter();
        statePutter = new StatePutter();
        triggerer = new TriggerFunction();
        uploader = new UploadFunction();
        downloader = new DownloadFunction();
        restart = new RestartFunction();
        stop = new StopFunction();
        start = new StartFunction();
        jcommander.addCommand("localrun", getLocalRunner());
        jcommander.addCommand("create", getCreater());
        jcommander.addCommand("delete", getDeleter());
        jcommander.addCommand("update", getUpdater());
        jcommander.addCommand("get", getGetter());
        jcommander.addCommand("restart", getRestarter());
        jcommander.addCommand("stop", getStopper());
        jcommander.addCommand("start", getStarter());
        // TODO depecreate getstatus
        jcommander.addCommand("status", getStatuser(), "getstatus");
        jcommander.addCommand("stats", getFunctionStats());
        jcommander.addCommand("list", getLister());
        jcommander.addCommand("querystate", getStateGetter());
        jcommander.addCommand("putstate", getStatePutter());
        jcommander.addCommand("trigger", getTriggerer());
        jcommander.addCommand("upload", getUploader());
        jcommander.addCommand("download", getDownloader());
    }

    @VisibleForTesting
    LocalRunner getLocalRunner() {
        return localRunner;
    }

    @VisibleForTesting
    CreateFunction getCreater() {
        return creater;
    }

    @VisibleForTesting
    DeleteFunction getDeleter() {
        return deleter;
    }

    @VisibleForTesting
    UpdateFunction getUpdater() {
        return updater;
    }

    @VisibleForTesting
    GetFunction getGetter() {
        return getter;
    }

    @VisibleForTesting
    GetFunctionStatus getStatuser() {
        return functionStatus;
    }

    @VisibleForTesting
    ListFunctions getLister() {
        return lister;
    }

    @VisibleForTesting
    StatePutter getStatePutter() {
        return statePutter;
    }

    @VisibleForTesting
    StateGetter getStateGetter() {
        return stateGetter;
    }

    @VisibleForTesting
    TriggerFunction getTriggerer() {
        return triggerer;
    }

    @VisibleForTesting
    UploadFunction getUploader() {
        return uploader;
    }

    @VisibleForTesting
    DownloadFunction getDownloader() {
        return downloader;
    }

    @VisibleForTesting
    RestartFunction getRestarter() {
        return restart;
    }

    @VisibleForTesting
    StopFunction getStopper() {
        return stop;
    }

    @VisibleForTesting
    StartFunction getStarter() {
        return start;
    }

    private void parseFullyQualifiedFunctionName(String fqfn, FunctionConfig functionConfig) {
        String[] args = fqfn.split("/");
        if (args.length != 3) {
            throw new ParameterException("Fully qualified function names (FQFNs) must "
                    + "be of the form tenant/namespace/name");
        } else {
            functionConfig.setTenant(args[0]);
            functionConfig.setNamespace(args[1]);
            functionConfig.setName(args[2]);
        }
    }

}