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 index 444b78f9ada..0fd61364280 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryMemTable.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryMemTable.java @@ -19,6 +19,8 @@ package org.apache.bookkeeper.bookie; +import io.netty.buffer.ByteBuf; +import io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Iterator; @@ -441,6 +443,26 @@ public EntryKeyValue getLastEntry(long ledgerId) throws IOException { return (EntryKeyValue) result; } + /** + Gets the lastAddConfirmed for a ledger from its highest available entry. + + Returns null if there are no entries for the given ledger + */ + public Long getLastAddConfirmed(long ledgerId) throws IOException { + Long lac = null; + EntryKeyValue ekv = getLastEntry(ledgerId); + if (ekv != null) { + ByteBuf bb = ekv.getValueAsByteBuffer(); + try { + bb.skipBytes(2 * Long.BYTES); // Skip ledger and entry id + lac = bb.readLong(); + } finally { + ReferenceCountUtil.release(bb); + } + } + return lac; + } + /** * Check if the entire heap usage for this EntryMemTable exceeds limit. */ diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java index f0d537f4d38..1c8345bcc98 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/IndexPersistenceMgr.java @@ -722,4 +722,10 @@ public LedgerCache.LedgerIndexMetadata readLedgerIndexMetadata(long ledgerId) th } } } + + @VisibleForTesting + void invalidateLedger(long ledgerId) { + this.writeFileInfoCache.invalidate(ledgerId); + this.readFileInfoCache.invalidate(ledgerId); + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java index fa1b724622d..57572bdb490 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java @@ -31,6 +31,7 @@ import static org.apache.bookkeeper.bookie.BookKeeperServerStats.STORAGE_SCRUB_PAGES_SCANNED; import static org.apache.bookkeeper.bookie.BookKeeperServerStats.STORAGE_SCRUB_PAGE_RETRIES; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; import com.google.common.util.concurrent.RateLimiter; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; @@ -360,17 +361,21 @@ public boolean entryExists(long ledgerId, long entryId) throws IOException { public long getLastAddConfirmed(long ledgerId) throws IOException { Long lac = ledgerCache.getLastAddConfirmed(ledgerId); if (lac == null) { - ByteBuf bb = getEntry(ledgerId, BookieProtocol.LAST_ADD_CONFIRMED); - if (null == bb) { - return BookieProtocol.INVALID_ENTRY_ID; - } else { - try { - bb.skipBytes(2 * Long.BYTES); // skip ledger & entry id - lac = bb.readLong(); - lac = ledgerCache.updateLastAddConfirmed(ledgerId, lac); - } finally { - ReferenceCountUtil.release(bb); - } + lac = getLastAddConfirmedFromLedgerEntry(ledgerId); + } + return lac; + } + + public long getLastAddConfirmedFromLedgerEntry(long ledgerId) throws IOException { + long lac = BookieProtocol.INVALID_ENTRY_ID; + ByteBuf bb = getEntry(ledgerId, BookieProtocol.LAST_ADD_CONFIRMED); + if (null != bb) { + try { + bb.skipBytes(2 * Long.BYTES); + lac = bb.readLong(); + lac = ledgerCache.updateLastAddConfirmed(ledgerId, lac); + } finally { + ReferenceCountUtil.release(bb); } } return lac; @@ -726,4 +731,9 @@ public void clearStorageStateFlag(StorageState flags) throws IOException { throw new UnsupportedOperationException( "Storage state flags only supported for DbLedgerStorage"); } + + @VisibleForTesting + void removeLedgerCacheEntry(long ledgerId) { + ((LedgerCacheImpl) ledgerCache).invalidateLedger(ledgerId); + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerCacheImpl.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerCacheImpl.java index 28a62c19e1e..390b263e481 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerCacheImpl.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerCacheImpl.java @@ -21,6 +21,7 @@ package org.apache.bookkeeper.bookie; +import com.google.common.annotations.VisibleForTesting; import io.netty.buffer.ByteBuf; import java.io.IOException; import java.util.Collections; @@ -233,4 +234,9 @@ public long nextLong() { } }; } + + @VisibleForTesting + void invalidateLedger(long ledgerId) { + indexPersistenceManager.invalidateLedger(ledgerId); + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java index bc3d586008c..425ad9fc806 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java @@ -240,7 +240,21 @@ public ByteBuf getEntry(long ledgerId, long entryId) throws IOException, BookieE @Override public long getLastAddConfirmed(long ledgerId) throws IOException { - return interleavedLedgerStorage.getLastAddConfirmed(ledgerId); + LedgerCache lc = interleavedLedgerStorage.getLedgerCache(); + Long lac = lc.getLastAddConfirmed(ledgerId); + if (null == lac) { + // If the last add confirmed has fallen out of the ledger cache, + // it may be in the memTable, and more importantly, it may be + // in the memTable but _not_ in ledger storage. + lac = memTable.getLastAddConfirmed(ledgerId); + if (null != lac) { + lc.updateLastAddConfirmed(ledgerId, lac); + } + } + if (null == lac) { + lac = interleavedLedgerStorage.getLastAddConfirmedFromLedgerEntry(ledgerId); + } + return lac; } @Override @@ -461,4 +475,9 @@ public void clearStorageStateFlag(StorageState flags) throws IOException { throw new UnsupportedOperationException( "Storage state flags only supported for DbLedgerStorage"); } + + @VisibleForTesting + void removeLedgerCacheEntry(long ledgerId) { + interleavedLedgerStorage.removeLedgerCacheEntry(ledgerId); + } } 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 index f2dbf643831..b9f5df2062e 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryMemTableTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryMemTableTest.java @@ -468,5 +468,34 @@ public void testAddSameEntries() throws IOException { assertEquals(memTable.kvmap.size(), 0); assertEquals(memTable.skipListSemaphore.availablePermits(), initialPermits); } + + @Test + public void testGetLastAddConfirmed() throws IOException { + final long ledgerId = 1; + final byte[] bytes = new byte[3 * Long.BYTES]; + final ByteBuffer bb = ByteBuffer.wrap(bytes); + + bb.putLong(0, 0); + bb.putLong(Long.BYTES, 0); + bb.putLong(2 * Long.BYTES, -1); + + memTable.addEntry(ledgerId, 0, bb.duplicate(), this); + + bb.putLong(2 * Long.BYTES, 0); + + memTable.addEntry(ledgerId, 1, bb.duplicate(), this); + + bb.putLong(2 * Long.BYTES, 1); + + memTable.addEntry(ledgerId, 2, bb.duplicate(), this); + + bb.putLong(2 * Long.BYTES, 2); + + memTable.addEntry(ledgerId, 3, bb.duplicate(), this); + + long lac = memTable.getLastAddConfirmed(ledgerId); + + assertEquals(2, lac); + } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorageTest.java index 9c75f9c3ad7..d6db69b4904 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorageTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorageTest.java @@ -25,6 +25,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertThrows; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -46,6 +47,7 @@ import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.conf.TestBKConfiguration; +import org.apache.bookkeeper.proto.BookieProtocol; import org.apache.bookkeeper.stats.NullStatsLogger; import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.test.TestStatsProvider; @@ -143,6 +145,11 @@ void checkEntry(long ledgerId, long entryId, long location) throws EntryLookupEx final long entriesPerWrite = 2; final long numOfLedgers = 5; + private void prepareLedger(final long ledgerId) throws IOException { + interleavedStorage.setMasterKey(ledgerId, ("ledger-" + ledgerId).getBytes()); + interleavedStorage.setFenced(ledgerId); + } + @Before public void setUp() throws Exception { File tmpDir = File.createTempFile("bkTest", ".dir"); @@ -167,8 +174,7 @@ public void setUp() throws Exception { for (long entryId = 0; entryId < numWrites; entryId++) { for (long ledgerId = 0; ledgerId < numOfLedgers; ledgerId++) { if (entryId == 0) { - interleavedStorage.setMasterKey(ledgerId, ("ledger-" + ledgerId).getBytes()); - interleavedStorage.setFenced(ledgerId); + this.prepareLedger(ledgerId); } ByteBuf entry = Unpooled.buffer(128); entry.writeLong(ledgerId); @@ -449,4 +455,58 @@ void printInfoLine(String s) { res = shell.run(new String[] { "localconsistencycheck" }); Assert.assertEquals(1, res); } + + @Test + public void testGetLastAddConfirmed() throws Exception { + final byte[] entry = new byte[Long.BYTES * 3]; + final ByteBuf bb = Unpooled.wrappedBuffer(entry); + + // Work off of a ledgerId that we haven't already inserted + final long ledgerId = this.numOfLedgers + 1; + final long entryId = 2; + final long lac = 1; + + this.prepareLedger(ledgerId); + bb.setLong(0, ledgerId); + bb.setLong(Long.BYTES, entryId); + bb.setLong(2 * Long.BYTES, lac); + + interleavedStorage.addEntry(bb); + + final long lac1 = interleavedStorage.getLastAddConfirmed(ledgerId); + final long lac2 = interleavedStorage.getLastAddConfirmed(ledgerId); + + interleavedStorage.removeLedgerCacheEntry(ledgerId); + + final long lac3 = interleavedStorage.getLastAddConfirmed(ledgerId); + + assertEquals(lac, lac1); + assertEquals(lac, lac2); + assertEquals(lac, lac3); + } + + @Test + public void testGetLastAddConfirmedFromLedgerEntry() throws Exception { + final byte[] entry = new byte[Long.BYTES * 3]; + final ByteBuf bb = Unpooled.wrappedBuffer(entry); + + // Work off of a ledgerId that we haven't already inserted + final long ledgerId = this.numOfLedgers + 2; + final long entryId = 2; + final long lac = 1; + + this.prepareLedger(ledgerId); + bb.setLong(0, ledgerId); + bb.setLong(Long.BYTES, entryId); + bb.setLong(2 * Long.BYTES, lac); + + assertThrows("Entry 0 not found", Bookie.NoEntryException.class, () -> { + interleavedStorage.getLastAddConfirmedFromLedgerEntry(ledgerId); + }); + + interleavedStorage.addEntry(bb); + + final long lac2 = interleavedStorage.getLastAddConfirmedFromLedgerEntry(ledgerId); + assertEquals(lac, lac2); + } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java index db83f096d95..44c819f0d3e 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/SortedLedgerStorageTest.java @@ -38,6 +38,7 @@ import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint; import org.apache.bookkeeper.conf.ServerConfiguration; import org.apache.bookkeeper.conf.TestBKConfiguration; +import org.apache.bookkeeper.proto.BookieProtocol; import org.apache.bookkeeper.test.TestStatsProvider; import org.apache.bookkeeper.util.DiskChecker; import org.junit.Before; @@ -191,4 +192,49 @@ public void testGetListOfEntriesOfLedgerAfterFlush() throws IOException { })); } } + + @Test + public void testLastAddConfirmedNotFlushedMissingInLedgerCache() throws Exception { + // Use a new ledgerId, so that prior tests don't interfere with this one + final long ledgerId = numOfLedgers + 1; + final byte[] writeKey = {0x00, 0x00, 0x00, 0x00}; + final byte[] payload = new byte[3 * Long.BYTES]; + ByteBuf bb = Unpooled.wrappedBuffer(payload); + HandleFactory hf = new HandleFactoryImpl(sortedLedgerStorage); + hf.getHandle(ledgerId, writeKey, false); + + bb.setLong(0, ledgerId); + + // Write first entry with invalid LAC + bb.setLong(Long.BYTES, 0); + bb.setLong(2 * Long.BYTES, BookieProtocol.INVALID_ENTRY_ID); + sortedLedgerStorage.addEntry(bb); + + // LAC entry should be both in the memTable and ledger cache + final long lac1 = sortedLedgerStorage.getLastAddConfirmed(ledgerId); + assertEquals(BookieProtocol.INVALID_ENTRY_ID, lac1); + + // But if the ledger cache is purged... + sortedLedgerStorage.removeLedgerCacheEntry(ledgerId); + + // Then the memTable should still be consulted + final long lac2 = sortedLedgerStorage.getLastAddConfirmed(ledgerId); + assertEquals(BookieProtocol.INVALID_ENTRY_ID, lac2); + + // With a new entry in sortedLedgerStorage... + bb.setLong(Long.BYTES, 1); + bb.setLong(2 * Long.BYTES, 0); + sortedLedgerStorage.addEntry(bb); + + // We should get an updated LAC from the ledger cache + final long lac3 = sortedLedgerStorage.getLastAddConfirmed(ledgerId); + assertEquals(0, lac3); + + // And once the ledger cache is purged again... + sortedLedgerStorage.removeLedgerCacheEntry(ledgerId); + + // We should read from the memTable + final long lac4 = sortedLedgerStorage.getLastAddConfirmed(ledgerId); + assertEquals(0, lac4); + } }