public static RuntimeOpts mergeRuntimeOpts(RuntimeOpts oriOpts, RuntimeOpts newOpts) {
        RuntimeOpts mergedOpts = oriOpts.partialDeepClone();
        if (mergedOpts.getExtraLabels() == null) {
            mergedOpts.setExtraLabels(new HashMap<>());
        }
        if (mergedOpts.getExtraAnnotations() == null) {
            mergedOpts.setExtraAnnotations(new HashMap<>());
        }
        if (mergedOpts.getNodeSelectorLabels() == null) {
            mergedOpts.setNodeSelectorLabels(new HashMap<>());
        }
        if (mergedOpts.getTolerations() == null) {
            mergedOpts.setTolerations(new ArrayList<>());
        }
        if (mergedOpts.getResourceRequirements() == null) {
            mergedOpts.setResourceRequirements(new V1ResourceRequirements());
        }

        if (!StringUtils.isEmpty(newOpts.getJobName())) {
            mergedOpts.setJobName(newOpts.getJobName());
        }
        if (!StringUtils.isEmpty(newOpts.getJobNamespace())) {
            mergedOpts.setJobNamespace(newOpts.getJobNamespace());
        }
        if (newOpts.getExtraLabels() != null && !newOpts.getExtraLabels().isEmpty()) {
            newOpts.getExtraLabels().forEach((key, labelsItem) -> {
                if (!mergedOpts.getExtraLabels().containsKey(key)) {
                    log.debug("extra label {} has been changed to {}", key, labelsItem);
                }
                mergedOpts.getExtraLabels().put(key, labelsItem);
            });
        }
        if (newOpts.getExtraAnnotations() != null && !newOpts.getExtraAnnotations().isEmpty()) {
            newOpts.getExtraAnnotations().forEach((key, annotationsItem) -> {
                if (!mergedOpts.getExtraAnnotations().containsKey(key)) {
                    log.debug("extra annotation {} has been changed to {}", key, annotationsItem);
                }
                mergedOpts.getExtraAnnotations().put(key, annotationsItem);
            });
        }
        if (newOpts.getNodeSelectorLabels() != null && !newOpts.getNodeSelectorLabels().isEmpty()) {
            newOpts.getNodeSelectorLabels().forEach((key, nodeSelectorItem) -> {
                if (!mergedOpts.getNodeSelectorLabels().containsKey(key)) {
                    log.debug("node selector label {} has been changed to {}", key, nodeSelectorItem);
                }
                mergedOpts.getNodeSelectorLabels().put(key, nodeSelectorItem);
            });
        }

        if (newOpts.getResourceRequirements() != null) {
            V1ResourceRequirements mergedResourcesRequirements = mergedOpts.getResourceRequirements();
            V1ResourceRequirements newResourcesRequirements = newOpts.getResourceRequirements();

            Map<String, Quantity> limits = newResourcesRequirements.getLimits();
            Map<String, Quantity> requests = newResourcesRequirements.getRequests();
            for (String resource : RESOURCES) {
                if (limits != null && limits.containsKey(resource)) {
                    mergedResourcesRequirements.putLimitsItem(resource, limits.get(resource));
                }
                if (requests != null && requests.containsKey(resource)) {
                    mergedResourcesRequirements.putRequestsItem(resource, requests.get(resource));
                }
            }
            mergedOpts.setResourceRequirements(mergedResourcesRequirements);
        }

        if (newOpts.getTolerations() != null && !newOpts.getTolerations().isEmpty()) {
            mergedOpts.getTolerations().addAll(newOpts.getTolerations());
        }
        return mergedOpts;
    }
@Test
    public void TestMergeRuntimeOpts() {
        Map<String, Object> configs = new Gson().fromJson(KubernetesRuntimeTest.createRuntimeCustomizerConfig(), HashMap.class);
        BasicKubernetesManifestCustomizer customizer = new BasicKubernetesManifestCustomizer();
        customizer.initialize(configs);
        BasicKubernetesManifestCustomizer.RuntimeOpts newOpts = new BasicKubernetesManifestCustomizer.RuntimeOpts();
        newOpts.setJobName("merged-name");
        newOpts.setTolerations(Collections.emptyList());
        V1Toleration toleration = new V1Toleration();
        toleration.setKey("merge-key");
        toleration.setEffect("NoSchedule");
        toleration.setOperator("Equal");
        toleration.setTolerationSeconds(6000L);
        newOpts.setTolerations(Collections.singletonList(toleration));
        V1ResourceRequirements resourceRequirements = new V1ResourceRequirements();
        resourceRequirements.putLimitsItem("cpu", new Quantity("20"));
        resourceRequirements.putLimitsItem("memory", new Quantity("10240"));
        newOpts.setResourceRequirements(resourceRequirements);
        newOpts.setNodeSelectorLabels(Collections.singletonMap("disktype", "ssd"));
        newOpts.setExtraAnnotations(Collections.singletonMap("functiontype", "sink"));
        newOpts.setExtraLabels(Collections.singletonMap("functiontype", "sink"));
        BasicKubernetesManifestCustomizer.RuntimeOpts mergedOpts = BasicKubernetesManifestCustomizer.mergeRuntimeOpts(
                customizer.getRuntimeOpts(), newOpts);

        assertEquals(mergedOpts.getJobName(), "merged-name");
        assertEquals(mergedOpts.getTolerations().size(), 2);
        assertEquals(mergedOpts.getExtraAnnotations().size(), 2);
        assertEquals(mergedOpts.getExtraLabels().size(), 2);
        assertEquals(mergedOpts.getNodeSelectorLabels().size(), 2);
        assertEquals(mergedOpts.getResourceRequirements().getLimits().get("cpu").getNumber().intValue(), 20);
        assertEquals(mergedOpts.getResourceRequirements().getLimits().get("memory").getNumber().intValue(), 10240);
    }