public static List<String> getGoInstanceCmd(InstanceConfig instanceConfig,
                                                AuthenticationConfig authConfig,
                                                String originalCodeFileName,
                                                String pulsarServiceUrl,
                                                String stateStorageServiceUrl,
                                                String pulsarWebServiceUrl,
                                                boolean k8sRuntime) throws IOException {
        final List<String> args = new LinkedList<>();
        GoInstanceConfig goInstanceConfig = new GoInstanceConfig();

        // pass the raw functino details directly so that we don't need to assemble the `instanceConf.funcDetails`
        // manually in Go instance
        String functionDetails =
                JsonFormat.printer().omittingInsignificantWhitespace().print(instanceConfig.getFunctionDetails());
        goInstanceConfig.setFunctionDetails(functionDetails);

        if (instanceConfig.getClusterName() != null) {
            goInstanceConfig.setClusterName(instanceConfig.getClusterName());
        }

        if (null != stateStorageServiceUrl) {
            goInstanceConfig.setStateStorageServiceUrl(stateStorageServiceUrl);
        }

        if (instanceConfig.isExposePulsarAdminClientEnabled() && StringUtils.isNotBlank(pulsarWebServiceUrl)) {
            goInstanceConfig.setPulsarWebServiceUrl(pulsarWebServiceUrl);
        }

        if (instanceConfig.getInstanceId() != 0) {
            goInstanceConfig.setInstanceID(instanceConfig.getInstanceId());
        }

        if (instanceConfig.getFunctionId() != null) {
            goInstanceConfig.setFuncID(instanceConfig.getFunctionId());
        }

        if (instanceConfig.getFunctionVersion() != null) {
            goInstanceConfig.setFuncVersion(instanceConfig.getFunctionVersion());
        }

        if (instanceConfig.getFunctionDetails().getAutoAck()) {
            goInstanceConfig.setAutoAck(instanceConfig.getFunctionDetails().getAutoAck());
        }

        if (instanceConfig.getFunctionDetails().getTenant() != null) {
            goInstanceConfig.setTenant(instanceConfig.getFunctionDetails().getTenant());
        }

        if (instanceConfig.getFunctionDetails().getNamespace() != null) {
            goInstanceConfig.setNameSpace(instanceConfig.getFunctionDetails().getNamespace());
        }

        if (instanceConfig.getFunctionDetails().getName() != null) {
            goInstanceConfig.setName(instanceConfig.getFunctionDetails().getName());
        }

        if (instanceConfig.getFunctionDetails().getLogTopic() != null) {
            goInstanceConfig.setLogTopic(instanceConfig.getFunctionDetails().getLogTopic());
        }
        if (instanceConfig.getFunctionDetails().getProcessingGuarantees() != null) {
            goInstanceConfig
                    .setProcessingGuarantees(instanceConfig.getFunctionDetails().getProcessingGuaranteesValue());
        }
        if (instanceConfig.getFunctionDetails().getRuntime() != null) {
            goInstanceConfig.setRuntime(instanceConfig.getFunctionDetails().getRuntimeValue());
        }
        if (instanceConfig.getFunctionDetails().getSecretsMap() != null) {
            goInstanceConfig.setSecretsMap(instanceConfig.getFunctionDetails().getSecretsMap());
        }
        if (instanceConfig.getFunctionDetails().getUserConfig() != null) {
            goInstanceConfig.setUserConfig(instanceConfig.getFunctionDetails().getUserConfig());
        }
        if (instanceConfig.getFunctionDetails().getParallelism() != 0) {
            goInstanceConfig.setParallelism(instanceConfig.getFunctionDetails().getParallelism());
        }

        if (authConfig != null) {
            if (isNotBlank(authConfig.getClientAuthenticationPlugin())
                    && isNotBlank(authConfig.getClientAuthenticationParameters())) {
                goInstanceConfig.setClientAuthenticationPlugin(authConfig.getClientAuthenticationPlugin());
                goInstanceConfig.setClientAuthenticationParameters(authConfig.getClientAuthenticationParameters());
            }
            goInstanceConfig.setTlsAllowInsecureConnection(
                    authConfig.isTlsAllowInsecureConnection());
            goInstanceConfig.setTlsHostnameVerificationEnable(
                    authConfig.isTlsHostnameVerificationEnable());
            if (isNotBlank(authConfig.getTlsTrustCertsFilePath())){
                goInstanceConfig.setTlsTrustCertsFilePath(
                        authConfig.getTlsTrustCertsFilePath());
            }

        }

        if (instanceConfig.getMaxBufferedTuples() != 0) {
            goInstanceConfig.setMaxBufTuples(instanceConfig.getMaxBufferedTuples());
        }

        if (pulsarServiceUrl != null) {
            goInstanceConfig.setPulsarServiceURL(pulsarServiceUrl);
        }
        if (instanceConfig.getFunctionDetails().getSource().getCleanupSubscription()) {
            goInstanceConfig
                    .setCleanupSubscription(instanceConfig.getFunctionDetails().getSource().getCleanupSubscription());
        }
        if (instanceConfig.getFunctionDetails().getSource().getSubscriptionName() != null) {
            goInstanceConfig.setSubscriptionName(instanceConfig.getFunctionDetails().getSource().getSubscriptionName());
        }
        goInstanceConfig.setSubscriptionPosition(
                instanceConfig.getFunctionDetails().getSource().getSubscriptionPosition().getNumber());

        if (instanceConfig.getFunctionDetails().getSource().getInputSpecsMap() != null) {
            Map<String, String> sourceInputSpecs = new HashMap<>();
            for (Map.Entry<String, Function.ConsumerSpec> entry :
                    instanceConfig.getFunctionDetails().getSource().getInputSpecsMap().entrySet()) {
                String topic = entry.getKey();
                Function.ConsumerSpec spec = entry.getValue();
                sourceInputSpecs.put(topic, JsonFormat.printer().omittingInsignificantWhitespace().print(spec));
                goInstanceConfig.setSourceSpecsTopic(topic);
            }
            goInstanceConfig.setSourceInputSpecs(sourceInputSpecs);
        }

        if (instanceConfig.getFunctionDetails().getSource().getTimeoutMs() != 0) {
            goInstanceConfig.setTimeoutMs(instanceConfig.getFunctionDetails().getSource().getTimeoutMs());
        }

        if (instanceConfig.getFunctionDetails().getSink().getTopic() != null) {
            goInstanceConfig.setSinkSpecsTopic(instanceConfig.getFunctionDetails().getSink().getTopic());
        }

        if (instanceConfig.getFunctionDetails().getResources().getCpu() != 0) {
            goInstanceConfig.setCpu(instanceConfig.getFunctionDetails().getResources().getCpu());
        }

        if (instanceConfig.getFunctionDetails().getResources().getRam() != 0) {
            goInstanceConfig.setRam(instanceConfig.getFunctionDetails().getResources().getRam());
        }

        if (instanceConfig.getFunctionDetails().getResources().getDisk() != 0) {
            goInstanceConfig.setDisk(instanceConfig.getFunctionDetails().getResources().getDisk());
        }

        if (instanceConfig.getFunctionDetails().getRetryDetails().getDeadLetterTopic() != null) {
            goInstanceConfig
                    .setDeadLetterTopic(instanceConfig.getFunctionDetails().getRetryDetails().getDeadLetterTopic());
        }

        if (instanceConfig.getFunctionDetails().getRetryDetails().getMaxMessageRetries() != 0) {
            goInstanceConfig
                    .setMaxMessageRetries(instanceConfig.getFunctionDetails().getRetryDetails().getMaxMessageRetries());
        }

        if (instanceConfig.hasValidMetricsPort()) {
            goInstanceConfig.setMetricsPort(instanceConfig.getMetricsPort());
        }

        goInstanceConfig.setKillAfterIdleMs(0);
        goInstanceConfig.setPort(instanceConfig.getPort());

        // Parse the contents of goInstanceConfig into json form string
        ObjectMapper objectMapper = ObjectMapperFactory.getMapper().getObjectMapper();
        String configContent = objectMapper.writeValueAsString(goInstanceConfig);

        args.add(originalCodeFileName);
        args.add("-instance-conf");
        if (k8sRuntime) {
            args.add("'" + configContent + "'");
        } else {
            args.add(configContent);
        }
        return args;
    }
@Test(dataProvider = "k8sRuntime")
    public void getGoInstanceCmd(boolean k8sRuntime) throws IOException {
        HashMap<String, String> goInstanceConfig;

        InstanceConfig instanceConfig = new InstanceConfig();
        instanceConfig.setClusterName("kluster");
        instanceConfig.setInstanceId(3000);
        instanceConfig.setFunctionId("func-7734");
        instanceConfig.setFunctionVersion("1.0.0");
        instanceConfig.setMaxBufferedTuples(5);
        instanceConfig.setPort(1337);
        instanceConfig.setMetricsPort(60000);

        AuthenticationConfig authConfig = AuthenticationConfig.builder()
                .clientAuthenticationPlugin("org.apache.pulsar.client.impl.auth.AuthenticationToken")
                .clientAuthenticationParameters("file:///secret/token.jwt")
                .tlsTrustCertsFilePath("/secret/ca.cert.pem")
                .tlsHostnameVerificationEnable(true)
                .tlsAllowInsecureConnection(false)
                .build();

        JSONObject userConfig = new JSONObject();
        userConfig.put("word-of-the-day", "der Weltschmerz");

        JSONObject secretsMap = new JSONObject();
        secretsMap.put("secret", "cake is a lie");

        Function.SourceSpec sources = Function.SourceSpec.newBuilder()
                .setCleanupSubscription(true)
                .setSubscriptionName("go-func-sub")
                .setTimeoutMs(500)
                .putInputSpecs("go-func-input", Function.ConsumerSpec.newBuilder().setIsRegexPattern(false).build())
                .build();

        Function.RetryDetails retryDetails = Function.RetryDetails.newBuilder()
                .setDeadLetterTopic("go-func-deadletter")
                .setMaxMessageRetries(1)
                .build();

        Function.Resources resources = Function.Resources.newBuilder()
                .setCpu(2)
                .setDisk(1024)
                .setRam(32)
                .build();

        Function.FunctionDetails functionDetails = Function.FunctionDetails.newBuilder()
                .setAutoAck(true)
                .setTenant("public")
                .setNamespace("default")
                .setName("go-func")
                .setLogTopic("go-func-log")
                .setProcessingGuarantees(Function.ProcessingGuarantees.ATLEAST_ONCE)
                .setRuntime(Function.FunctionDetails.Runtime.GO)
                .setSecretsMap(secretsMap.toJSONString())
                .setParallelism(1)
                .setSource(sources)
                .setRetryDetails(retryDetails)
                .setResources(resources)
                .setUserConfig(userConfig.toJSONString())
                .build();

        instanceConfig.setFunctionDetails(functionDetails);
        instanceConfig.setExposePulsarAdminClientEnabled(true);

        List<String> commands = RuntimeUtils.getGoInstanceCmd(instanceConfig, authConfig, "config", "pulsar://localhost:6650", "bk://localhost:4181",  "http://localhost:8080", k8sRuntime);
        if (k8sRuntime) {
            goInstanceConfig = new ObjectMapper().readValue(commands.get(2).replaceAll("^\'|\'$", ""), HashMap.class);
        } else {
            goInstanceConfig = new ObjectMapper().readValue(commands.get(2), HashMap.class);
        }
        Assert.assertEquals(commands.toArray().length, 3);
        Assert.assertEquals(commands.get(0), "config");
        Assert.assertEquals(commands.get(1), "-instance-conf");
        Assert.assertEquals(goInstanceConfig.get("maxBufTuples"), 5);
        Assert.assertEquals(goInstanceConfig.get("maxMessageRetries"), 1);
        Assert.assertEquals(goInstanceConfig.get("killAfterIdleMs"), 0);
        Assert.assertEquals(goInstanceConfig.get("parallelism"), 1);
        Assert.assertEquals(goInstanceConfig.get("className"), "");
        Assert.assertEquals(goInstanceConfig.get("sourceSpecsTopic"), "go-func-input");
        Assert.assertEquals(goInstanceConfig.get("secretsMap"), secretsMap.toString());
        Assert.assertEquals(goInstanceConfig.get("sourceSchemaType"), "");
        Assert.assertEquals(goInstanceConfig.get("sinkSpecsTopic"), "");
        Assert.assertEquals(goInstanceConfig.get("clusterName"), "kluster");
        Assert.assertEquals(goInstanceConfig.get("nameSpace"), "default");
        Assert.assertEquals(goInstanceConfig.get("receiverQueueSize"), 0);
        Assert.assertEquals(goInstanceConfig.get("tenant"), "public");
        Assert.assertEquals(goInstanceConfig.get("ram"), 32);
        Assert.assertEquals(goInstanceConfig.get("logTopic"), "go-func-log");
        Assert.assertEquals(goInstanceConfig.get("processingGuarantees"), 0);
        Assert.assertEquals(goInstanceConfig.get("autoAck"), true);
        Assert.assertEquals(goInstanceConfig.get("regexPatternSubscription"), false);
        Assert.assertEquals(goInstanceConfig.get("pulsarServiceURL"), "pulsar://localhost:6650");
        Assert.assertEquals(goInstanceConfig.get("stateStorageServiceUrl"), "bk://localhost:4181");
        Assert.assertEquals(goInstanceConfig.get("pulsarWebServiceUrl"), "http://localhost:8080");
        Assert.assertEquals(goInstanceConfig.get("runtime"), 3);
        Assert.assertEquals(goInstanceConfig.get("cpu"), 2.0);
        Assert.assertEquals(goInstanceConfig.get("funcID"), "func-7734");
        Assert.assertEquals(goInstanceConfig.get("funcVersion"), "1.0.0");
        Assert.assertEquals(goInstanceConfig.get("disk"), 1024);
        Assert.assertEquals(goInstanceConfig.get("instanceID"), 3000);
        Assert.assertEquals(goInstanceConfig.get("cleanupSubscription"), true);
        Assert.assertEquals(goInstanceConfig.get("port"), 1337);
        Assert.assertEquals(goInstanceConfig.get("subscriptionType"), 0);
        Assert.assertEquals(goInstanceConfig.get("timeoutMs"), 500);
        Assert.assertEquals(goInstanceConfig.get("subscriptionName"), "go-func-sub");
        Assert.assertEquals(goInstanceConfig.get("name"), "go-func");
        Assert.assertEquals(goInstanceConfig.get("expectedHealthCheckInterval"), 0);
        Assert.assertEquals(goInstanceConfig.get("deadLetterTopic"), "go-func-deadletter");
        Assert.assertEquals(goInstanceConfig.get("userConfig"), userConfig.toString());
        Assert.assertEquals(goInstanceConfig.get("metricsPort"), 60000);
        Assert.assertEquals(goInstanceConfig.get("clientAuthenticationPlugin"), "org.apache.pulsar.client.impl.auth.AuthenticationToken");
        Assert.assertEquals(goInstanceConfig.get("clientAuthenticationParameters"), "file:///secret/token.jwt");
        Assert.assertEquals(goInstanceConfig.get("tlsTrustCertsFilePath"), "/secret/ca.cert.pem");
        Assert.assertEquals(goInstanceConfig.get("tlsHostnameVerificationEnable"), true);
        Assert.assertEquals(goInstanceConfig.get("tlsAllowInsecureConnection"), false);
    }