private void addKubernetesProperties(Resource resource, FlinkConf flinkConf) {
    String clusterId = kubernetesClusterId(resource);
    flinkConf.putToOptions(FlinkConfKeys.KUBERNETES_CLUSTER_ID, clusterId);

    String[] labels = {
      "amoro.optimizing-group:" + resource.getGroupName(),
      "amoro.optimizer-implement:flink-native-kubernetes",
      "amoro.optimizer-id:" + resource.getResourceId()
    };
    String resourceLabel = Joiner.on(',').join(labels);
    flinkConf.putToOptions(FlinkConfKeys.KUBERNETES_TASKMANAGER_LABELS, resourceLabel);
    flinkConf.putToOptions(FlinkConfKeys.KUBERNETES_JOBMANAGER_LABELS, resourceLabel);
  }
Copy from flink 1.20 GlobalConfiguration.flatten Utils

@param config
@param keyPrefix
@return
amoro-ams/src/main/java/org/apache/amoro/server/manager/FlinkOptimizerContainer.java