code: public static void mergeHadoopConf(Configuration hadoopConfig) {
  
  		// we have to load the global configuration here, because the HadoopInputFormatBase does not
  		// have access to a Flink configuration object
  		org.apache.flink.configuration.Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration();
  
  		Configuration hadoopConf =
  			org.apache.flink.api.java.hadoop.mapred.utils.HadoopUtils.getHadoopConfiguration(flinkConfiguration);
  
  		for (Map.Entry<String, String> e : hadoopConf) {
  			if (hadoopConfig.get(e.getKey()) == null) {
  				hadoopConfig.set(e.getKey(), e.getValue());
  			}
  		}
  	}
docs: Merge HadoopConfiguration into Configuration. This is necessary for the HDFS configuration.