Fix file system on spark push job

LiveIntent · May 22, 2024 · f23dcf8 · f23dcf8
1 parent 2720a83
commit f23dcf8
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 4 deletions.
diff --git a/...main/java/com/linkedin/venice/hadoop/spark/input/hdfs/VeniceHdfsInputPartitionReader.java b/...main/java/com/linkedin/venice/hadoop/spark/input/hdfs/VeniceHdfsInputPartitionReader.java
@@ -39,11 +39,9 @@ protected VeniceRecordIterator createRecordIterator(VeniceProperties jobConfig,
       throw new VeniceException("Expected VeniceHdfsInputPartition");
     }
     VeniceHdfsInputPartition inputPartition = (VeniceHdfsInputPartition) partition;
-
-    Configuration configuration = new Configuration();
     FileSystem fs;
     try {
-      fs = FileSystem.get(configuration);
+      fs = inputPartition.getFilePath().getFileSystem(new Configuration());
     } catch (IOException e) {
       throw new VeniceException("Unable to get a FileSystem", e);
     }

diff --git a/...sh-job/src/main/java/com/linkedin/venice/hadoop/spark/input/hdfs/VeniceHdfsInputScan.java b/...sh-job/src/main/java/com/linkedin/venice/hadoop/spark/input/hdfs/VeniceHdfsInputScan.java
@@ -32,7 +32,7 @@ public VeniceHdfsInputScan(VeniceProperties jobConfig) {
   public InputPartition[] planInputPartitions() {
     try {
       Path inputDirPath = new Path(jobConfig.getString(INPUT_PATH_PROP));
-      FileSystem fs = FileSystem.get(new Configuration());
+      FileSystem fs = inputDirPath.getFileSystem(new Configuration());
       List<VeniceHdfsInputPartition> inputPartitionList = new ArrayList<>();
       // For now, we create 1 file as 1 InputPartition. This is not the most ideal, because Avro allows splitting files
       // to a smaller granularity using sync markers. We can explore later if we feel we need that optimization.