Fix core dumps triggered by rocksdb compacting when shutdown bk

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndex.java

@@ -25,6 +25,7 @@
 import java.io.IOException;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.TimeUnit;
 import org.apache.bookkeeper.bookie.Bookie;
 import org.apache.bookkeeper.bookie.EntryLocation;
@@ -48,7 +49,7 @@ public class EntryLocationIndex implements Closeable {
     private final KeyValueStorage locationsDb;
     private final ConcurrentLongHashSet deletedLedgers = ConcurrentLongHashSet.newBuilder().build();
     private final EntryLocationIndexStats stats;
-    private boolean isCompacting;
+    private final AtomicBoolean compacting = new AtomicBoolean(false);
 
     public EntryLocationIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
             StatsLogger stats) throws IOException {
@@ -203,15 +204,21 @@ public String getEntryLocationDBPath() {
 
     public void compact() throws IOException {
         try {
-            isCompacting = true;
+            if (!compacting.compareAndSet(false, true)) {
+                return;
+            }
             locationsDb.compact();
         } finally {
-            isCompacting = false;
+            compacting.set(false);
         }
     }
 
     public boolean isCompacting() {
-        return isCompacting;
+        return compacting.get();
+    }
+
+    public boolean compareAndSetCompacting(boolean expectedValue, boolean newValue) {
+        return compacting.compareAndSet(expectedValue, newValue);
     }
 
     public void removeOffsetFromDeletedLedgers() throws IOException {

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java

@@ -347,6 +347,9 @@ public void shutdown() throws InterruptedException {
         try {
             flush();
 
+            while (!entryLocationIndex.compareAndSetCompacting(false, true)) {
+                Thread.sleep(100);
+            }
             gcThread.shutdown();
             entryLogger.close();