[FLINK-22294] Fix throw IllegalArgumentException when getNumFiles with partitions on different hdfs nameservices

Author: zuston

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveSourceFileEnumerator.java

@@ -110,13 +110,10 @@ public static InputSplit[] createMRSplits(
     public static int getNumFiles(List<HiveTablePartition> partitions, JobConf jobConf)
             throws IOException {
         int numFiles = 0;
-        FileSystem fs = null;
         for (HiveTablePartition partition : partitions) {
             StorageDescriptor sd = partition.getStorageDescriptor();
             org.apache.hadoop.fs.Path inputPath = new org.apache.hadoop.fs.Path(sd.getLocation());
-            if (fs == null) {
-                fs = inputPath.getFileSystem(jobConf);
-            }
+            FileSystem fs = inputPath.getFileSystem(jobConf);
             // it's possible a partition exists in metastore but the data has been removed
             if (!fs.exists(inputPath)) {
                 continue;

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java

@@ -345,13 +345,10 @@ public InputSplitAssigner getInputSplitAssigner(HiveTableInputSplit[] inputSplit
 
     public int getNumFiles() throws IOException {
         int numFiles = 0;
-        FileSystem fs = null;
         for (HiveTablePartition partition : partitions) {
             StorageDescriptor sd = partition.getStorageDescriptor();
             Path inputPath = new Path(sd.getLocation());
-            if (fs == null) {
-                fs = inputPath.getFileSystem(jobConf.conf());
-            }
+            FileSystem fs = inputPath.getFileSystem(jobConf.conf());
             // it's possible a partition exists in metastore but the data has been removed
             if (!fs.exists(inputPath)) {
                 continue;