Merge pull request #957 from mjanczykowski/bugfix/AMQ-9199

[AMQ-9199] Fixed race condition in creating store directory

activemq-broker/src/main/java/org/apache/activemq/util/IOHelper.java

@@ -328,6 +328,10 @@ public final class IOHelper {
 
         } else {
             if (!dir.mkdirs()) {
+                if ( dir.exists() && dir.isDirectory() ) {
+                    // Directory created in parallel
+                    return;
+                }
                 throw new IOException("Failed to create directory '" + dir + "'");
             }
         }

activemq-kahadb-store/src/test/java/org/apache/activemq/store/kahadb/MessageDatabaseTest.java

@@ -28,6 +28,8 @@ import org.junit.rules.TemporaryFolder;
 import java.io.File;
 import java.io.IOException;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.activemq.store.kahadb.disk.journal.Journal.DEFAULT_MAX_WRITE_BATCH_SIZE;
@@ -114,4 +116,42 @@ public class MessageDatabaseTest {
             kaha.stop();
         }
     }
+
+    @Test
+    public void testKahaStartAndSizeCreatingStoreDirectoryConcurrently() throws Exception {
+        // given mkdirs() will execute in parallel
+        final CountDownLatch countDownLatch = new CountDownLatch(2);
+
+        class ConcurrentMkdirsFile extends File {
+
+            public ConcurrentMkdirsFile(File parent, String child) {
+                super(parent, child);
+            }
+
+            @Override
+            public boolean mkdirs() {
+                countDownLatch.countDown();
+                try {
+                    countDownLatch.await(3, TimeUnit.SECONDS);
+                    return super.mkdirs();
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+
+        // and KahaDBStore is configured
+        KahaDBStore kaha = new KahaDBStore();
+        kaha.setDirectory(new ConcurrentMkdirsFile(temporaryFolder.getRoot(), "kaha3"));
+
+        // when both start() and size() are performed concurrently
+        final Future<Long> size = Executors.newSingleThreadExecutor()
+                .submit(kaha::size);
+        kaha.start();
+
+        // then KahaDB should successfully start
+        assertTrue(kaha.isStarted());
+        assertTrue(size.get() > 0);
+    }
+
 }