Release semaphore when addEntry accepts the same entries (apache#2832)

sijie · Oct 16, 2021 · a9b576d · a9b576d
1 parent 148bf22
commit a9b576d
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 0 deletions.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryMemTable.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryMemTable.java
@@ -320,6 +320,9 @@ public long addEntry(long ledgerId, long entryId, final ByteBuffer entry, final
             try {
                 EntryKeyValue toAdd = cloneWithAllocator(ledgerId, entryId, entry);
                 size = internalAdd(toAdd);
+                if (size == 0) {
+                    skipListSemaphore.release(len);
+                }
             } finally {
                 this.lock.readLock().unlock();
             }

diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryMemTableTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryMemTableTest.java
@@ -451,5 +451,25 @@ public void run() {
             assertEquals("listOfEntries should be sorted", Long.valueOf(i + 1), listOfEntries.get(i));
         }
     }
+
+    @Test
+    public void testAddSameEntries() throws IOException {
+        final long ledgerId = 1;
+        final long entryId = 1;
+        final int size = 10;
+        final byte[] bytes = new byte[size];
+        final int initialPermits = memTable.skipListSemaphore.availablePermits();
+
+        for (int i = 0; i < 5; i++) {
+            memTable.addEntry(ledgerId, entryId, ByteBuffer.wrap(bytes), this);
+            assertEquals(memTable.kvmap.size(), 1);
+            assertEquals(memTable.skipListSemaphore.availablePermits(), initialPermits - size);
+        }
+
+        memTable.snapshot(Checkpoint.MAX);
+        memTable.flush(this);
+        assertEquals(memTable.kvmap.size(), 0);
+        assertEquals(memTable.skipListSemaphore.availablePermits(), initialPermits);
+    }
 }