public static YarnConfiguration getYarnConfiguration(
            org.apache.flink.configuration.Configuration flinkConfig) {
        final YarnConfiguration yarnConfig = new YarnConfiguration();

        for (String key : flinkConfig.keySet()) {
            for (String prefix : FLINK_CONFIG_PREFIXES) {
                if (key.startsWith(prefix)) {
                    String newKey = key.substring("flink.".length());
                    String value = flinkConfig.getString(key, null);
                    yarnConfig.set(newKey, value);
                    LOG.debug(
                            "Adding Flink config entry for {} as {}={} to Yarn config",
                            key,
                            newKey,
                            value);
                }
            }
        }

        return yarnConfig;
    }
@Test
    void testGetYarnConfiguration() {
        final String flinkPrefix = "flink.yarn.";
        final String yarnPrefix = "yarn.";

        final String k1 = "brooklyn";
        final String v1 = "nets";

        final String k2 = "golden.state";
        final String v2 = "warriors";

        final String k3 = "miami";
        final String v3 = "heat";

        final Configuration flinkConfig = new Configuration();
        flinkConfig.setString(flinkPrefix + k1, v1);
        flinkConfig.setString(flinkPrefix + k2, v2);
        flinkConfig.setString(k3, v3);

        final YarnConfiguration yarnConfig = Utils.getYarnConfiguration(flinkConfig);

        assertThat(yarnConfig.get(yarnPrefix + k1, null)).isEqualTo(v1);
        assertThat(yarnConfig.get(yarnPrefix + k2, null)).isEqualTo(v2);
        assertThat(yarnConfig.get(yarnPrefix + k3)).isNull();
    }