[iceberg] branch master updated: Use ConcurrentHashMap in ResolvingFileIO (#8292)

[et...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

etudenhoefner pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/master by this push:
     new 9ec6084cc2 Use ConcurrentHashMap in ResolvingFileIO (#8292)
9ec6084cc2 is described below

commit 9ec6084cc21217dc27bb5db870fb0d7b607e3ad1
Author: Xianyang Liu <li...@hotmail.com>
AuthorDate: Fri Aug 11 16:38:52 2023 +0800

    Use ConcurrentHashMap in ResolvingFileIO (#8292)
---
 .../org/apache/iceberg/io/ResolvingFileIO.java     | 84 ++++++++++------------
 1 file changed, 38 insertions(+), 46 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java b/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java
index d927c0a164..22f17ec4db 100644
--- a/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java
+++ b/core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java
@@ -50,7 +50,7 @@ public class ResolvingFileIO implements FileIO, HadoopConfigurable {
           "s3n", S3_FILE_IO_IMPL,
           "gs", GCS_FILE_IO_IMPL);
 
-  private final Map<String, FileIO> ioInstances = Maps.newHashMap();
+  private final Map<String, FileIO> ioInstances = Maps.newConcurrentMap();
   private final AtomicBoolean isClosed = new AtomicBoolean(false);
   private final transient StackTraceElement[] createStack;
   private SerializableMap<String, String> properties;
@@ -102,10 +102,8 @@ public class ResolvingFileIO implements FileIO, HadoopConfigurable {
     if (isClosed.compareAndSet(false, true)) {
       List<FileIO> instances = Lists.newArrayList();
 
-      synchronized (ioInstances) {
-        instances.addAll(ioInstances.values());
-        ioInstances.clear();
-      }
+      instances.addAll(ioInstances.values());
+      ioInstances.clear();
 
       for (FileIO io : instances) {
         io.close();
@@ -134,7 +132,7 @@ public class ResolvingFileIO implements FileIO, HadoopConfigurable {
     FileIO io = ioInstances.get(impl);
     if (io != null) {
       if (io instanceof HadoopConfigurable && ((HadoopConfigurable) io).getConf() == null) {
-        synchronized (ioInstances) {
+        synchronized (io) {
           if (((HadoopConfigurable) io).getConf() == null) {
             // re-apply the config in case it's null after Kryo serialization
             ((HadoopConfigurable) io).setConf(hadoopConf.get());
@@ -145,49 +143,43 @@ public class ResolvingFileIO implements FileIO, HadoopConfigurable {
       return io;
     }
 
-    synchronized (ioInstances) {
-      // double check while holding the lock
-      io = ioInstances.get(impl);
-      if (io != null) {
-        return io;
-      }
+    return ioInstances.computeIfAbsent(
+        impl,
+        key -> {
+          Configuration conf = hadoopConf.get();
 
-      Configuration conf = hadoopConf.get();
-
-      try {
-        Map<String, String> props = Maps.newHashMap(properties);
-        // ResolvingFileIO is keeping track of the creation stacktrace, so no need to do the same in
-        // S3FileIO.
-        props.put("init-creation-stacktrace", "false");
-        io = CatalogUtil.loadFileIO(impl, props, conf);
-      } catch (IllegalArgumentException e) {
-        if (impl.equals(FALLBACK_IMPL)) {
-          // no implementation to fall back to, throw the exception
-          throw e;
-        } else {
-          // couldn't load the normal class, fall back to HadoopFileIO
-          LOG.warn(
-              "Failed to load FileIO implementation: {}, falling back to {}",
-              impl,
-              FALLBACK_IMPL,
-              e);
           try {
-            io = CatalogUtil.loadFileIO(FALLBACK_IMPL, properties, conf);
-          } catch (IllegalArgumentException suppressed) {
-            LOG.warn(
-                "Failed to load FileIO implementation: {} (fallback)", FALLBACK_IMPL, suppressed);
-            // both attempts failed, throw the original exception with the later exception
-            // suppressed
-            e.addSuppressed(suppressed);
-            throw e;
+            Map<String, String> props = Maps.newHashMap(properties);
+            // ResolvingFileIO is keeping track of the creation stacktrace, so no need to do the
+            // same in S3FileIO.
+            props.put("init-creation-stacktrace", "false");
+            return CatalogUtil.loadFileIO(key, props, conf);
+          } catch (IllegalArgumentException e) {
+            if (key.equals(FALLBACK_IMPL)) {
+              // no implementation to fall back to, throw the exception
+              throw e;
+            } else {
+              // couldn't load the normal class, fall back to HadoopFileIO
+              LOG.warn(
+                  "Failed to load FileIO implementation: {}, falling back to {}",
+                  key,
+                  FALLBACK_IMPL,
+                  e);
+              try {
+                return CatalogUtil.loadFileIO(FALLBACK_IMPL, properties, conf);
+              } catch (IllegalArgumentException suppressed) {
+                LOG.warn(
+                    "Failed to load FileIO implementation: {} (fallback)",
+                    FALLBACK_IMPL,
+                    suppressed);
+                // both attempts failed, throw the original exception with the later exception
+                // suppressed
+                e.addSuppressed(suppressed);
+                throw e;
+              }
+            }
           }
-        }
-      }
-
-      ioInstances.put(impl, io);
-    }
-
-    return io;
+        });
   }
 
   private static String implFromLocation(String location) {