HBASE-28222 Leak in ExportSnapshot during verifySnapshot on S3A (#5554)

Revert "HBASE-12819 ExportSnapshot doesn't close FileSystem instances"

This reverts commit ee32eeb

Signed-off-by: Wellington Chevreuil <wchevreuil@apache.org>

Author: bbeaudreault

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/snapshot/ExportSnapshot.java

@@ -59,7 +59,6 @@
 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.io.BytesWritable;
-import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapreduce.InputFormat;
@@ -209,14 +208,12 @@ public void setup(Context context) throws IOException {
       outputArchive = new Path(outputRoot, HConstants.HFILE_ARCHIVE_DIRECTORY);
 
       try {
-        srcConf.setBoolean("fs." + inputRoot.toUri().getScheme() + ".impl.disable.cache", true);
         inputFs = FileSystem.get(inputRoot.toUri(), srcConf);
       } catch (IOException e) {
         throw new IOException("Could not get the input FileSystem with root=" + inputRoot, e);
       }
 
       try {
-        destConf.setBoolean("fs." + outputRoot.toUri().getScheme() + ".impl.disable.cache", true);
         outputFs = FileSystem.get(outputRoot.toUri(), destConf);
       } catch (IOException e) {
         throw new IOException("Could not get the output FileSystem with root=" + outputRoot, e);
@@ -238,12 +235,6 @@ public void setup(Context context) throws IOException {
       }
     }
 
-    @Override
-    protected void cleanup(Context context) {
-      IOUtils.closeStream(inputFs);
-      IOUtils.closeStream(outputFs);
-    }
-
     @Override
     public void map(BytesWritable key, NullWritable value, Context context)
       throws InterruptedException, IOException {
@@ -987,10 +978,8 @@ public int doWork() throws IOException {
     }
 
     Configuration srcConf = HBaseConfiguration.createClusterConf(conf, null, CONF_SOURCE_PREFIX);
-    srcConf.setBoolean("fs." + inputRoot.toUri().getScheme() + ".impl.disable.cache", true);
     FileSystem inputFs = FileSystem.get(inputRoot.toUri(), srcConf);
     Configuration destConf = HBaseConfiguration.createClusterConf(conf, null, CONF_DEST_PREFIX);
-    destConf.setBoolean("fs." + outputRoot.toUri().getScheme() + ".impl.disable.cache", true);
     FileSystem outputFs = FileSystem.get(outputRoot.toUri(), destConf);
     boolean skipTmp = conf.getBoolean(CONF_SKIP_TMP, false)
       || conf.get(SnapshotDescriptionUtils.SNAPSHOT_WORKING_DIR) != null;
@@ -1146,9 +1135,6 @@ public int doWork() throws IOException {
       }
       outputFs.delete(outputSnapshotDir, true);
       return 1;
-    } finally {
-      IOUtils.closeStream(inputFs);
-      IOUtils.closeStream(outputFs);
     }
   }