optimize and ut

Author: houxiaoyu

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

@@ -20,6 +20,7 @@
  */
 package org.apache.bookkeeper.bookie.storage.ldb;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Iterables;
 import java.io.Closeable;
 import java.io.IOException;
@@ -49,7 +50,8 @@ public class EntryLocationIndex implements Closeable {
     private final KeyValueStorage locationsDb;
     private final ConcurrentLongHashSet deletedLedgers = ConcurrentLongHashSet.newBuilder().build();
     private final EntryLocationIndexStats stats;
-    private final AtomicBoolean compacting = new AtomicBoolean(false);
+    @VisibleForTesting
+    final AtomicBoolean compacting = new AtomicBoolean(false);
 
     public EntryLocationIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
             StatsLogger stats) throws IOException {
@@ -68,22 +70,19 @@ public EntryLocationIndex(ServerConfiguration conf, KeyValueStorageFactory stora
 
     @Override
     public void close() throws IOException {
-        long start = System.currentTimeMillis();
         log.info("Closing EntryLocationIndex");
         while (!compacting.compareAndSet(false, true)) {
             // Wait till the locationsDb stops compacting
-            if ((System.currentTimeMillis() - start) % 1000 == 0) {
-                log.info("Waiting the locationsDb stops compacting");
-            }
+            log.info("Waiting the locationsDb stops compacting");
             try {
-                Thread.sleep(100);
+                Thread.sleep(1000);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 throw new IOException(e);
             }
         }
         locationsDb.close();
-        log.info("Closed EntryLocationIndex cost: {} mills", System.currentTimeMillis() - start);
+        log.info("Closed EntryLocationIndex");
     }
 
     public long getLocation(long ledgerId, long entryId) throws IOException {

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndexTest.java

@@ -21,14 +21,19 @@
 package org.apache.bookkeeper.bookie.storage.ldb;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.test.TestStatsProvider;
+import org.awaitility.Awaitility;
 import org.junit.Test;
+import org.junit.jupiter.api.Timeout;
 
 /**
  * Unit test for {@link EntryLocationIndex}.
@@ -231,4 +236,39 @@ public void testEntryIndexLookupLatencyStats() throws IOException {
         assertEquals(1, lookupEntryLocationOpStats.getFailureCount());
         assertEquals(1, lookupEntryLocationOpStats.getSuccessCount());
     }
+
+    @Test
+    @Timeout(30)
+    public void testClose() throws Exception {
+        File tmpDir = File.createTempFile("bkTest", ".dir");
+        tmpDir.delete();
+        tmpDir.mkdir();
+        tmpDir.deleteOnExit();
+
+        EntryLocationIndex idx = new EntryLocationIndex(serverConfiguration, KeyValueStorageRocksDB.factory,
+                tmpDir.getAbsolutePath(), NullStatsLogger.INSTANCE);
+
+        // mock EntryLocationIndex is compacting
+        idx.compacting.set(true);
+        AtomicBoolean flag = new AtomicBoolean(false);
+        AtomicLong costMills = new AtomicLong(0);
+        new Thread(() -> {
+            try {
+                long start = System.currentTimeMillis();
+                idx.close();
+                costMills.set(System.currentTimeMillis() - start);
+                flag.set(true);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }).start();
+        long sleepMills = 10_000;
+        Thread.sleep(sleepMills);
+        assertFalse(flag.get());
+
+        // mock EntryLocationIndex finish compacting
+        idx.compacting.set(false);
+        Awaitility.await().untilAsserted(() -> assertTrue(flag.get()));
+        assertTrue(costMills.get() >= sleepMills);
+    }
 }