HDDS-4313. Create thread-local instance of FileSystem in HadoopFsGenerator

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/BaseFreonGenerator.java

@@ -162,11 +162,20 @@ private void taskLoop(TaskProvider provider) {
 
       //in case of an other failed test, we shouldn't execute more tasks.
       if (counter >= testNo || (!failAtEnd && failureCounter.get() > 0)) {
-        return;
+        break;
       }
 
       tryNextTask(provider, counter);
     }
+
+    taskLoopCompleted();
+  }
+
+  /**
+   * Provides a way to clean up per-thread resources.
+   */
+  protected void taskLoopCompleted() {
+    // no-op
   }
 
   /**

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/freon/HadoopFsGenerator.java

@@ -16,6 +16,8 @@
  */
 package org.apache.hadoop.ozone.freon;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.URI;
 import java.util.concurrent.Callable;
 
@@ -26,8 +28,6 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 
 import com.codahale.metrics.Timer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.Option;
 
@@ -43,9 +43,6 @@
 public class HadoopFsGenerator extends BaseFreonGenerator
     implements Callable<Void> {
 
-  private static final Logger LOG =
-      LoggerFactory.getLogger(HadoopFsGenerator.class);
-
   @Option(names = {"--path"},
       description = "Hadoop FS file system path",
       defaultValue = "o3fs://bucket1.vol1")
@@ -70,16 +67,26 @@ public class HadoopFsGenerator extends BaseFreonGenerator
 
   private Timer timer;
 
-  private FileSystem fileSystem;
+  private OzoneConfiguration configuration;
+  private URI uri;
+  private final ThreadLocal<FileSystem> threadLocalFileSystem =
+      ThreadLocal.withInitial(this::createFS);
 
   @Override
   public Void call() throws Exception {
-
     init();
 
-    OzoneConfiguration configuration = createOzoneConfiguration();
+    configuration = createOzoneConfiguration();
+    uri = URI.create(rootPath);
+    String disableCacheName = String.format("fs.%s.impl.disable.cache",
+        uri.getScheme());
+    print("Disabling FS cache: " + disableCacheName);
+    configuration.setBoolean(disableCacheName, true);
 
-    fileSystem = FileSystem.get(URI.create(rootPath), configuration);
+    Path file = new Path(rootPath + "/" + generateObjectName(0));
+    try (FileSystem fileSystem = threadLocalFileSystem.get()) {
+      fileSystem.mkdirs(file.getParent());
+    }
 
     contentGenerator =
         new ContentGenerator(fileSize, bufferSize, copyBufferSize);
@@ -93,7 +100,7 @@ public Void call() throws Exception {
 
   private void createFile(long counter) throws Exception {
     Path file = new Path(rootPath + "/" + generateObjectName(counter));
-    fileSystem.mkdirs(file.getParent());
+    FileSystem fileSystem = threadLocalFileSystem.get();
 
     timer.time(() -> {
       try (FSDataOutputStream output = fileSystem.create(file)) {
@@ -102,4 +109,22 @@ private void createFile(long counter) throws Exception {
       return null;
     });
   }
+
+  private FileSystem createFS() {
+    try {
+      return FileSystem.get(uri, configuration);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  @Override
+  protected void taskLoopCompleted() {
+    FileSystem fileSystem = threadLocalFileSystem.get();
+    try {
+      fileSystem.close();
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
 }