Resolve the problem that YarnClusterClientFactory cannot load yarn configurations

YarnClusterClientFactory. getClusterDescriptor method   Unable to load the configuration for yarn .  The reason is that it is called HadoopUtils.getHadoopConfiguration and this method only loading HDFS configuration.

The call chain looks like this: YarnClusterClientFactory#getClusterDescriptor->Utils#getYarnAndHadoopConfiguration->  HadoopUtils.getHadoopConfiguration --> Hadooputills#addHadoopConfIfFound 
However, the HadoopUtils#addHadoopConfIfFound method does not load yarn configuration information

First,YarnClusterClientFactory#getClusterDescriptor This method call Utils.getYarnAndHadoopConfiguration method

private YarnClusterDescriptor getClusterDescriptor(Configuration configuration)  
Unknown macro: { final YarnClient yarnClient = YarnClient.createYarnClient(); final YarnConfiguration yarnConfiguration = Utils.getYarnAndHadoopConfiguration(configuration); yarnClient.init(yarnConfiguration); yarnClient.start(); return new YarnClusterDescriptor( configuration, yarnConfiguration, yarnClient, YarnClientYarnClusterInformationRetriever.create(yarnClient), false); }

It then calls Utils# getYarnAndHadoopConfiguration method, in the call HadoopUtils# getHadoopConfiguration methods will only loading the Hadoop configuration unable to load the configuration for Yarn.

    public static YarnConfiguration getYarnAndHadoopConfiguration(
            org.apache.flink.configuration.Configuration flinkConfig) 
Unknown macro: {         final YarnConfiguration yarnConfig = getYarnConfiguration(flinkConfig);         yarnConfig.addResource(HadoopUtils.getHadoopConfiguration(flinkConfig));         return yarnConfig;     }

Then in HadoopUtils. GetHadoopConfiguration methods this Approach in the 3 will through HadoopUtils# addHadoopConfIfFound method to load the configuration file

    public static Configuration getHadoopConfiguration(
    
    。。。。。
    
// Approach 3: HADOOP_CONF_DIR environment variable
        String hadoopConfDir = System.getenv("HADOOP_CONF_DIR");
        if (hadoopConfDir != null) {
            LOG.debug("Searching Hadoop configuration files in HADOOP_CONF_DIR: {}", hadoopConfDir);
            foundHadoopConfiguration =
                    addHadoopConfIfFound(result, hadoopConfDir) || foundHadoopConfiguration;
        }
        。。。。。
}

Finally, it calls the Hadooputills#addHadoopConfIfFound  method, which loads only the core-site and hdfs-site configuration but not the yarn-site configuration

private static boolean addHadoopConfIfFound(

Configuration configuration, String possibleHadoopConfPath)
Unknown macro: { boolean foundHadoopConfiguration = false; if (new File(possibleHadoopConfPath).exists()) Unknown macro}
if (new File(possibleHadoopConfPath + "/hdfs-site.xml").exists())
Unknown macro: { configuration.addResource( new org.apache.hadoop.fs.Path(possibleHadoopConfPath + "/hdfs-site.xml")); LOG.debug( "Adding " + possibleHadoopConfPath + "/hdfs-site.xml to hadoop configuration"); foundHadoopConfiguration = true; }
}

return foundHadoopConfiguration;

}