fix corruption due to data race when concurrently calling

cpp/src/arrow/filesystem/s3fs.cc

@@ -2898,12 +2898,13 @@ struct AwsInstance {
     if (is_finalized_.load()) {
       return Status::Invalid("Attempt to initialize S3 after it has been finalized");
     }
-    if (!is_initialized_.exchange(true)) {
-      // Not already initialized
+    bool newly_initialized = false;
+    std::call_once(initialize_flag_, [&]() {
       DoInitialize(options);
-      return true;
-    }
-    return false;
+      is_initialized_.exchange(true);
+      newly_initialized = true;
+    });
+    return newly_initialized;
   }
 
   bool IsInitialized() { return !is_finalized_ && is_initialized_; }
@@ -2979,6 +2980,7 @@ struct AwsInstance {
   Aws::SDKOptions aws_options_;
   std::atomic<bool> is_initialized_;
   std::atomic<bool> is_finalized_;
+  std::once_flag initialize_flag_;
 };
 
 AwsInstance* GetAwsInstance() {

python/pyarrow/tests/test_fs.py

@@ -1904,3 +1904,22 @@ def test_s3_finalize_region_resolver():
             resolve_s3_region('voltrondata-labs-datasets')
         """
     subprocess.check_call([sys.executable, "-c", code])
+
+@pytest.mark.s3
+def test_concurrent_fs_init():
+    code = """if 1:
+        import threading
+        import pytest
+        from pyarrow.fs import (FileSystem, S3FileSystem
+                                ensure_s3_initialized, finalize_s3) 
+        threads = []
+        for i in range(4):
+            thread = threading.Thread(target = lambda: FileSystem.from_uri('s3://mf-nwp-models/README.txt'))
+            threads.append(thread)
+            thread.start()
+
+        for thread in threads:
+            thread.join()
+
+        finalize_s3()
+    """