[FLINK-20710][hive] Hive reading fail when splits on different FileSystem nameservices

This closes #14452

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

@@ -62,13 +62,10 @@ public class HiveSourceFileEnumerator implements FileEnumerator {
 			List<HiveTablePartition> partitions,
 			JobConf jobConf) throws IOException {
 		List<HiveSourceSplit> hiveSplits = new ArrayList<>();
-		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

@@ -285,13 +285,10 @@ public class HiveTableInputFormat extends HadoopInputFormatCommonBase<RowData, H
 			JobConf jobConf) throws IOException {
 		List<HiveTableInputSplit> hiveSplits = new ArrayList<>();
 		int splitNum = 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);
-			}
+			FileSystem fs = inputPath.getFileSystem(jobConf);
 			// it's possible a partition exists in metastore but the data has been removed
 			if (!fs.exists(inputPath)) {
 				continue;