public static FunctionConfig convertFromDetails(FunctionDetails functionDetails) {
        functionDetails = validateFunctionDetails(functionDetails);
        FunctionConfig functionConfig = new FunctionConfig();
        functionConfig.setTenant(functionDetails.getTenant());
        functionConfig.setNamespace(functionDetails.getNamespace());
        functionConfig.setName(functionDetails.getName());
        functionConfig.setParallelism(functionDetails.getParallelism());
        functionConfig.setProcessingGuarantees(
                FunctionCommon.convertProcessingGuarantee(functionDetails.getProcessingGuarantees()));
        Map<String, ConsumerConfig> consumerConfigMap = new HashMap<>();
        for (Map.Entry<String, Function.ConsumerSpec> input : functionDetails.getSource().getInputSpecsMap()
                .entrySet()) {
            ConsumerConfig consumerConfig = new ConsumerConfig();
            if (isNotEmpty(input.getValue().getSerdeClassName())) {
                consumerConfig.setSerdeClassName(input.getValue().getSerdeClassName());
            }
            if (isNotEmpty(input.getValue().getSchemaType())) {
                consumerConfig.setSchemaType(input.getValue().getSchemaType());
            }
            if (input.getValue().hasReceiverQueueSize()) {
                consumerConfig.setReceiverQueueSize(input.getValue().getReceiverQueueSize().getValue());
            }
            if (input.getValue().hasCryptoSpec()) {
                consumerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(input.getValue().getCryptoSpec()));
            }
            consumerConfig.setRegexPattern(input.getValue().getIsRegexPattern());
            consumerConfig.setSchemaProperties(input.getValue().getSchemaPropertiesMap());
            consumerConfig.setPoolMessages(input.getValue().getPoolMessages());
            consumerConfigMap.put(input.getKey(), consumerConfig);
        }
        functionConfig.setInputSpecs(consumerConfigMap);
        if (!isEmpty(functionDetails.getSource().getSubscriptionName())) {
            functionConfig.setSubName(functionDetails.getSource().getSubscriptionName());
        }
        functionConfig.setRetainOrdering(functionDetails.getRetainOrdering());
        functionConfig.setRetainKeyOrdering(functionDetails.getRetainKeyOrdering());

        functionConfig.setCleanupSubscription(functionDetails.getSource().getCleanupSubscription());
        functionConfig.setAutoAck(functionDetails.getAutoAck());

        // Set subscription position
        functionConfig.setSubscriptionPosition(
                convertFromFunctionDetailsSubscriptionPosition(functionDetails.getSource().getSubscriptionPosition()));

        if (functionDetails.getSource().getTimeoutMs() != 0) {
            functionConfig.setTimeoutMs(functionDetails.getSource().getTimeoutMs());
        }
        if (!isEmpty(functionDetails.getSink().getTopic())) {
            functionConfig.setOutput(functionDetails.getSink().getTopic());
        }
        if (!isEmpty(functionDetails.getSink().getSerDeClassName())) {
            functionConfig.setOutputSerdeClassName(functionDetails.getSink().getSerDeClassName());
        }
        if (!isEmpty(functionDetails.getSink().getSchemaType())) {
            functionConfig.setOutputSchemaType(functionDetails.getSink().getSchemaType());
        }
        if (functionDetails.getSink().getProducerSpec() != null) {
            functionConfig.setProducerConfig(
                    convertProducerSpecToProducerConfig(functionDetails.getSink().getProducerSpec()));
        }
        if (!isEmpty(functionDetails.getLogTopic())) {
            functionConfig.setLogTopic(functionDetails.getLogTopic());
        }
        if (functionDetails.getSink().getForwardSourceMessageProperty()) {
            functionConfig.setForwardSourceMessageProperty(functionDetails.getSink().getForwardSourceMessageProperty());
        }
        functionConfig.setRuntime(FunctionCommon.convertRuntime(functionDetails.getRuntime()));
        if (functionDetails.hasRetryDetails()) {
            functionConfig.setMaxMessageRetries(functionDetails.getRetryDetails().getMaxMessageRetries());
            if (!isEmpty(functionDetails.getRetryDetails().getDeadLetterTopic())) {
                functionConfig.setDeadLetterTopic(functionDetails.getRetryDetails().getDeadLetterTopic());
            }
        }
        Map<String, Object> userConfig;
        if (!isEmpty(functionDetails.getUserConfig())) {
            Type type = new TypeToken<Map<String, Object>>() {
            }.getType();
            userConfig = new Gson().fromJson(functionDetails.getUserConfig(), type);
        } else {
            userConfig = new HashMap<>();
        }
        if (userConfig.containsKey(WindowConfig.WINDOW_CONFIG_KEY)) {
            WindowConfig windowConfig = new Gson().fromJson(
                    (new Gson().toJson(userConfig.get(WindowConfig.WINDOW_CONFIG_KEY))),
                    WindowConfig.class);
            userConfig.remove(WindowConfig.WINDOW_CONFIG_KEY);
            if (windowConfig.getProcessingGuarantees() != null) {
                functionConfig.setProcessingGuarantees(
                        FunctionConfig.ProcessingGuarantees.valueOf(windowConfig.getProcessingGuarantees().name()));
            }
            functionConfig.setClassName(windowConfig.getActualWindowFunctionClassName());
            functionConfig.setWindowConfig(windowConfig);
        } else {
            functionConfig.setClassName(functionDetails.getClassName());
        }
        functionConfig.setUserConfig(userConfig);

        if (!isEmpty(functionDetails.getSecretsMap())) {
            Type type = new TypeToken<Map<String, Object>>() {
            }.getType();
            Map<String, Object> secretsMap = new Gson().fromJson(functionDetails.getSecretsMap(), type);
            functionConfig.setSecrets(secretsMap);
        }

        if (functionDetails.hasResources()) {
            Resources resources = new Resources();
            resources.setCpu(functionDetails.getResources().getCpu());
            resources.setRam(functionDetails.getResources().getRam());
            resources.setDisk(functionDetails.getResources().getDisk());
            functionConfig.setResources(resources);
        }

        if (!isEmpty(functionDetails.getRuntimeFlags())) {
            functionConfig.setRuntimeFlags(functionDetails.getRuntimeFlags());
        }

        if (!isEmpty(functionDetails.getCustomRuntimeOptions())) {
            functionConfig.setCustomRuntimeOptions(functionDetails.getCustomRuntimeOptions());
        }

        return functionConfig;
    }
@Test
    public void testFunctionConfigConvertFromDetails() {
        String name = "test1";
        String namespace = "ns1";
        String tenant = "tenant1";
        String classname = getClass().getName();
        int parallelism = 3;
        Map<String, String> userConfig = new HashMap<>();
        userConfig.put("key1", "val1");
        Function.ProcessingGuarantees processingGuarantees = Function.ProcessingGuarantees.EFFECTIVELY_ONCE;
        Function.FunctionDetails.Runtime runtime = Function.FunctionDetails.Runtime.JAVA;
        Function.SinkSpec sinkSpec = Function.SinkSpec.newBuilder().setTopic("sinkTopic1").build();
        Map<String, Function.ConsumerSpec> consumerSpecMap = new HashMap<>();
        consumerSpecMap.put("sourceTopic1", Function.ConsumerSpec.newBuilder()
                .setSchemaType(JSONSchema.class.getName()).build());
        Function.SourceSpec sourceSpec = Function.SourceSpec.newBuilder()
                .putAllInputSpecs(consumerSpecMap)
                .setSubscriptionType(Function.SubscriptionType.FAILOVER)
                .setCleanupSubscription(true)
                .build();
        boolean autoAck = true;
        String logTopic = "log-topic1";
        Function.Resources resources = Function.Resources.newBuilder().setCpu(1.5).setDisk(1024 * 20).setRam(1024 * 10).build();
        String packageUrl = "http://package.url";
        Map<String, String> secretsMap = new HashMap<>();
        secretsMap.put("secretConfigKey1", "secretConfigVal1");
        Function.RetryDetails retryDetails = Function.RetryDetails.newBuilder().setDeadLetterTopic("dead-letter-1").build();

        Function.FunctionDetails functionDetails = Function.FunctionDetails
                .newBuilder()
                .setNamespace(namespace)
                .setTenant(tenant)
                .setName(name)
                .setClassName(classname)
                .setParallelism(parallelism)
                .setUserConfig(new Gson().toJson(userConfig))
                .setProcessingGuarantees(processingGuarantees)
                .setRuntime(runtime)
                .setSink(sinkSpec)
                .setSource(sourceSpec)
                .setAutoAck(autoAck)
                .setLogTopic(logTopic)
                .setResources(resources)
                .setPackageUrl(packageUrl)
                .setSecretsMap(new Gson().toJson(secretsMap))
                .setRetryDetails(retryDetails)
                .build();

        FunctionConfig functionConfig = FunctionConfigUtils.convertFromDetails(functionDetails);

        assertEquals(functionConfig.getTenant(), tenant);
        assertEquals(functionConfig.getNamespace(), namespace);
        assertEquals(functionConfig.getName(), name);
        assertEquals(functionConfig.getClassName(), classname);
        assertEquals(functionConfig.getLogTopic(), logTopic);
        assertEquals((Object) functionConfig.getResources().getCpu(), resources.getCpu());
        assertEquals(functionConfig.getResources().getDisk().longValue(), resources.getDisk());
        assertEquals(functionConfig.getResources().getRam().longValue(), resources.getRam());
        assertEquals(functionConfig.getOutput(), sinkSpec.getTopic());
        assertEquals(functionConfig.getInputSpecs().keySet(), sourceSpec.getInputSpecsMap().keySet());
        assertEquals(functionConfig.getCleanupSubscription().booleanValue(), sourceSpec.getCleanupSubscription());
    }