locations) throws IOException {
+ // Trigger a flush to have all the entries being compacted in the db storage
+ flush();
+
+ entryLocationIndex.updateLocations(locations);
+ }
+
+ @Override
+ public EntryLogger getEntryLogger() {
+ return entryLogger;
+ }
+
+ @Override
+ public long getLastAddConfirmed(long ledgerId) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Observable waitForLastAddConfirmedUpdate(long ledgerId, long previoisLAC, Observer observer)
+ throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setExplicitlac(long ledgerId, ByteBuf lac) throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ByteBuf getExplicitLac(long ledgerId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void flushEntriesLocationsIndex() throws IOException {
+ // No-op. Location index is already flushed in updateEntriesLocations() call
+ }
+
+ /**
+ * Add an already existing ledger to the index.
+ *
+ * This method is only used as a tool to help the migration from
+ * InterleaveLedgerStorage to DbLedgerStorage
+ *
+ * @param ledgerId
+ * the ledger id
+ * @param entries
+ * a map of entryId -> location
+ * @return the number of
+ */
+ public long addLedgerToIndex(long ledgerId, boolean isFenced, byte[] masterKey,
+ Iterable> entries) throws Exception {
+ LedgerData ledgerData = LedgerData.newBuilder().setExists(true).setFenced(isFenced)
+ .setMasterKey(ByteString.copyFrom(masterKey)).build();
+ ledgerIndex.set(ledgerId, ledgerData);
+ AtomicLong numberOfEntries = new AtomicLong();
+
+ // Iterate over all the entries pages
+ Batch batch = entryLocationIndex.newBatch();
+ entries.forEach(map -> {
+ map.forEach((entryId, location) -> {
+ try {
+ entryLocationIndex.addLocation(batch, ledgerId, entryId, location);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ numberOfEntries.incrementAndGet();
+ });
+ });
+
+ batch.flush();
+ batch.close();
+
+ return numberOfEntries.get();
+ }
+
+ @Override
+ public void registerLedgerDeletionListener(LedgerDeletionListener listener) {
+ ledgerDeletionListeners.add(listener);
+ }
+
+ public EntryLocationIndex getEntryLocationIndex() {
+ return entryLocationIndex;
+ }
+
+ private void recordSuccessfulEvent(OpStatsLogger logger, long startTimeNanos) {
+ logger.registerSuccessfulEvent(MathUtils.elapsedNanos(startTimeNanos), TimeUnit.NANOSECONDS);
+ }
+
+ private void recordFailedEvent(OpStatsLogger logger, long startTimeNanos) {
+ logger.registerFailedEvent(MathUtils.elapsedNanos(startTimeNanos), TimeUnit.NANOSECONDS);
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(DbLedgerStorage.class);
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndex.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndex.java
new file mode 100644
index 00000000000..53e37a263cf
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndex.java
@@ -0,0 +1,232 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import com.google.common.collect.Iterables;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.EntryLocation;
+import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorage.Batch;
+import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorageFactory.DbConfigType;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.stats.Gauge;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.util.collections.ConcurrentLongHashSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Maintains an index of the entry locations in the EntryLogger.
+ *
+ * For each ledger multiple entries are stored in the same "record", represented
+ * by the {@link LedgerIndexPage} class.
+ */
+public class EntryLocationIndex implements Closeable {
+
+ private final KeyValueStorage locationsDb;
+ private final ConcurrentLongHashSet deletedLedgers = new ConcurrentLongHashSet();
+
+ private StatsLogger stats;
+
+ public EntryLocationIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
+ StatsLogger stats) throws IOException {
+ String locationsDbPath = FileSystems.getDefault().getPath(basePath, "locations").toFile().toString();
+ locationsDb = storageFactory.newKeyValueStorage(locationsDbPath, DbConfigType.Huge, conf);
+
+ this.stats = stats;
+ registerStats();
+ }
+
+ public void registerStats() {
+ stats.registerGauge("entries-count", new Gauge() {
+ @Override
+ public Long getDefaultValue() {
+ return 0L;
+ }
+
+ @Override
+ public Long getSample() {
+ try {
+ return locationsDb.count();
+ } catch (IOException e) {
+ return -1L;
+ }
+ }
+ });
+ }
+
+ @Override
+ public void close() throws IOException {
+ locationsDb.close();
+ }
+
+ public long getLocation(long ledgerId, long entryId) throws IOException {
+ LongPairWrapper key = LongPairWrapper.get(ledgerId, entryId);
+ LongWrapper value = LongWrapper.get();
+
+ try {
+ if (locationsDb.get(key.array, value.array) < 0) {
+ if (log.isDebugEnabled()) {
+ log.debug("Entry not found {}@{} in db index", ledgerId, entryId);
+ }
+ return 0;
+ }
+
+ return value.getValue();
+ } finally {
+ key.recycle();
+ value.recycle();
+ }
+ }
+
+ public long getLastEntryInLedger(long ledgerId) throws IOException {
+ if (deletedLedgers.contains(ledgerId)) {
+ // Ledger already deleted
+ return -1;
+ }
+
+ return getLastEntryInLedgerInternal(ledgerId);
+ }
+
+ private long getLastEntryInLedgerInternal(long ledgerId) throws IOException {
+ LongPairWrapper maxEntryId = LongPairWrapper.get(ledgerId, Long.MAX_VALUE);
+
+ // Search the last entry in storage
+ Entry entry = locationsDb.getFloor(maxEntryId.array);
+ maxEntryId.recycle();
+
+ if (entry == null) {
+ throw new Bookie.NoEntryException(ledgerId, -1);
+ } else {
+ long foundLedgerId = ArrayUtil.getLong(entry.getKey(), 0);
+ long lastEntryId = ArrayUtil.getLong(entry.getKey(), 8);
+
+ if (foundLedgerId == ledgerId) {
+ if (log.isDebugEnabled()) {
+ log.debug("Found last page in storage db for ledger {} - last entry: {}", ledgerId, lastEntryId);
+ }
+ return lastEntryId;
+ } else {
+ throw new Bookie.NoEntryException(ledgerId, -1);
+ }
+ }
+ }
+
+ public void addLocation(long ledgerId, long entryId, long location) throws IOException {
+ Batch batch = locationsDb.newBatch();
+ addLocation(batch, ledgerId, entryId, location);
+ batch.flush();
+ batch.close();
+ }
+
+ public Batch newBatch() {
+ return locationsDb.newBatch();
+ }
+
+ public void addLocation(Batch batch, long ledgerId, long entryId, long location) throws IOException {
+ LongPairWrapper key = LongPairWrapper.get(ledgerId, entryId);
+ LongWrapper value = LongWrapper.get(location);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Add location - ledger: {} -- entry: {} -- location: {}", ledgerId, entryId, location);
+ }
+
+ try {
+ batch.put(key.array, value.array);
+ } finally {
+ key.recycle();
+ value.recycle();
+ }
+ }
+
+ public void updateLocations(Iterable newLocations) throws IOException {
+ if (log.isDebugEnabled()) {
+ log.debug("Update locations -- {}", Iterables.size(newLocations));
+ }
+
+ Batch batch = newBatch();
+ // Update all the ledger index pages with the new locations
+ for (EntryLocation e : newLocations) {
+ if (log.isDebugEnabled()) {
+ log.debug("Update location - ledger: {} -- entry: {}", e.ledger, e.entry);
+ }
+
+ addLocation(batch, e.ledger, e.entry, e.location);
+ }
+
+ batch.flush();
+ batch.close();
+ }
+
+ public void delete(long ledgerId) throws IOException {
+ // We need to find all the LedgerIndexPage records belonging to one specific
+ // ledgers
+ deletedLedgers.add(ledgerId);
+ }
+
+ public void removeOffsetFromDeletedLedgers() throws IOException {
+ LongPairWrapper firstKeyWrapper = LongPairWrapper.get(-1, -1);
+ LongPairWrapper lastKeyWrapper = LongPairWrapper.get(-1, -1);
+ LongPairWrapper keyToDelete = LongPairWrapper.get(-1, -1);
+
+ Set ledgersToDelete = deletedLedgers.items();
+
+ if (ledgersToDelete.isEmpty()) {
+ return;
+ }
+
+ log.info("Deleting indexes for ledgers: {}", ledgersToDelete);
+ Batch batch = locationsDb.newBatch();
+
+ try {
+ for (long ledgerId : ledgersToDelete) {
+ if (log.isDebugEnabled()) {
+ log.debug("Deleting indexes from ledger {}", ledgerId);
+ }
+
+ firstKeyWrapper.set(ledgerId, 0);
+ lastKeyWrapper.set(ledgerId, Long.MAX_VALUE);
+
+ batch.deleteRange(firstKeyWrapper.array, lastKeyWrapper.array);
+ }
+
+ batch.flush();
+
+ // Removed from pending set
+ for (long ledgerId : ledgersToDelete) {
+ deletedLedgers.remove(ledgerId);
+ }
+ } finally {
+ firstKeyWrapper.recycle();
+ lastKeyWrapper.recycle();
+ keyToDelete.recycle();
+ batch.close();
+ }
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(EntryLocationIndex.class);
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java
new file mode 100644
index 00000000000..04bf32dba71
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java
@@ -0,0 +1,262 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+import com.google.protobuf.ByteString;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.util.AbstractMap.SimpleEntry;
+import java.util.Arrays;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorageDataFormats.LedgerData;
+import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorage.CloseableIterator;
+import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorageFactory.DbConfigType;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.stats.Gauge;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Maintains an index for the ledgers metadata.
+ *
+ * The key is the ledgerId and the value is the {@link LedgerData} content.
+ */
+public class LedgerMetadataIndex implements Closeable {
+ // Contains all ledgers stored in the bookie
+ private final ConcurrentLongHashMap ledgers;
+ private final AtomicInteger ledgersCount;
+
+ private final KeyValueStorage ledgersDb;
+ private StatsLogger stats;
+
+ // Holds ledger modifications applied in memory map, and pending to be flushed on db
+ private final ConcurrentLinkedQueue> pendingLedgersUpdates;
+
+ // Holds ledger ids that were delete from memory map, and pending to be flushed on db
+ private final ConcurrentLinkedQueue pendingDeletedLedgers;
+
+ public LedgerMetadataIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
+ StatsLogger stats) throws IOException {
+ String ledgersPath = FileSystems.getDefault().getPath(basePath, "ledgers").toFile().toString();
+ ledgersDb = storageFactory.newKeyValueStorage(ledgersPath, DbConfigType.Small, conf);
+
+ ledgers = new ConcurrentLongHashMap<>();
+ ledgersCount = new AtomicInteger();
+
+ // Read all ledgers from db
+ CloseableIterator> iterator = ledgersDb.iterator();
+ try {
+ while (iterator.hasNext()) {
+ Entry entry = iterator.next();
+ long ledgerId = ArrayUtil.getLong(entry.getKey(), 0);
+ LedgerData ledgerData = LedgerData.parseFrom(entry.getValue());
+ ledgers.put(ledgerId, ledgerData);
+ ledgersCount.incrementAndGet();
+ }
+ } finally {
+ iterator.close();
+ }
+
+ this.pendingLedgersUpdates = new ConcurrentLinkedQueue>();
+ this.pendingDeletedLedgers = new ConcurrentLinkedQueue();
+
+ this.stats = stats;
+ registerStats();
+ }
+
+ public void registerStats() {
+ stats.registerGauge("ledgers-count", new Gauge() {
+ @Override
+ public Long getDefaultValue() {
+ return 0L;
+ }
+
+ @Override
+ public Long getSample() {
+ return (long) ledgersCount.get();
+ }
+ });
+ }
+
+ @Override
+ public void close() throws IOException {
+ ledgersDb.close();
+ }
+
+ public LedgerData get(long ledgerId) throws IOException {
+ LedgerData ledgerData = ledgers.get(ledgerId);
+ if (ledgerData == null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Ledger not found {}", ledgerId);
+ }
+ throw new Bookie.NoLedgerException(ledgerId);
+ }
+
+ return ledgerData;
+ }
+
+ public void set(long ledgerId, LedgerData ledgerData) throws IOException {
+ ledgerData = LedgerData.newBuilder(ledgerData).setExists(true).build();
+
+ if (ledgers.put(ledgerId, ledgerData) == null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Added new ledger {}", ledgerId);
+ }
+ ledgersCount.incrementAndGet();
+ }
+
+ pendingLedgersUpdates.add(new SimpleEntry(ledgerId, ledgerData));
+ pendingDeletedLedgers.remove(ledgerId);
+ }
+
+ public void delete(long ledgerId) throws IOException {
+ if (ledgers.remove(ledgerId) != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Removed ledger {}", ledgerId);
+ }
+ ledgersCount.decrementAndGet();
+ }
+
+ pendingDeletedLedgers.add(ledgerId);
+ pendingLedgersUpdates.removeIf(e -> e.getKey() == ledgerId);
+ }
+
+ public Iterable getActiveLedgersInRange(final long firstLedgerId, final long lastLedgerId)
+ throws IOException {
+ return Iterables.filter(ledgers.keys(), new Predicate() {
+ @Override
+ public boolean apply(Long ledgerId) {
+ return ledgerId >= firstLedgerId && ledgerId < lastLedgerId;
+ }
+ });
+ }
+
+ public boolean setFenced(long ledgerId) throws IOException {
+ LedgerData ledgerData = get(ledgerId);
+ if (ledgerData.getFenced()) {
+ return false;
+ }
+
+ LedgerData newLedgerData = LedgerData.newBuilder(ledgerData).setFenced(true).build();
+
+ if (ledgers.put(ledgerId, newLedgerData) == null) {
+ // Ledger had been deleted
+ if (log.isDebugEnabled()) {
+ log.debug("Re-inserted fenced ledger {}", ledgerId);
+ }
+ ledgersCount.incrementAndGet();
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Set fenced ledger {}", ledgerId);
+ }
+ }
+
+ pendingLedgersUpdates.add(new SimpleEntry(ledgerId, newLedgerData));
+ pendingDeletedLedgers.remove(ledgerId);
+ return true;
+ }
+
+ public void setMasterKey(long ledgerId, byte[] masterKey) throws IOException {
+ LedgerData ledgerData = ledgers.get(ledgerId);
+ if (ledgerData == null) {
+ // New ledger inserted
+ ledgerData = LedgerData.newBuilder().setExists(true).setFenced(false)
+ .setMasterKey(ByteString.copyFrom(masterKey)).build();
+ if (log.isDebugEnabled()) {
+ log.debug("Inserting new ledger {}", ledgerId);
+ }
+ } else {
+ byte[] storedMasterKey = ledgerData.getMasterKey().toByteArray();
+ if (ArrayUtil.isArrayAllZeros(storedMasterKey)) {
+ // update master key of the ledger
+ ledgerData = LedgerData.newBuilder(ledgerData).setMasterKey(ByteString.copyFrom(masterKey)).build();
+ if (log.isDebugEnabled()) {
+ log.debug("Replace old master key {} with new master key {}", storedMasterKey, masterKey);
+ }
+ } else if (!Arrays.equals(storedMasterKey, masterKey) && !ArrayUtil.isArrayAllZeros(masterKey)) {
+ log.warn("Ledger {} masterKey in db can only be set once.", ledgerId);
+ throw new IOException(BookieException.create(BookieException.Code.IllegalOpException));
+ }
+ }
+
+ if (ledgers.put(ledgerId, ledgerData) == null) {
+ ledgersCount.incrementAndGet();
+ }
+
+ pendingLedgersUpdates.add(new SimpleEntry(ledgerId, ledgerData));
+ pendingDeletedLedgers.remove(ledgerId);
+ }
+
+ /**
+ * Flushes all pending changes.
+ */
+ public void flush() throws IOException {
+ LongWrapper key = LongWrapper.get();
+
+ int updatedLedgers = 0;
+ while (!pendingLedgersUpdates.isEmpty()) {
+ Entry entry = pendingLedgersUpdates.poll();
+ key.set(entry.getKey());
+ byte[] value = entry.getValue().toByteArray();
+ ledgersDb.put(key.array, value);
+ ++updatedLedgers;
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("Persisting updates to {} ledgers", updatedLedgers);
+ }
+
+ ledgersDb.sync();
+ key.recycle();
+ }
+
+ public void removeDeletedLedgers() throws IOException {
+ LongWrapper key = LongWrapper.get();
+
+ int deletedLedgers = 0;
+ while (!pendingDeletedLedgers.isEmpty()) {
+ long ledgerId = pendingDeletedLedgers.poll();
+ key.set(ledgerId);
+ ledgersDb.delete(key.array);
+ deletedLedgers++;
+ }
+
+ if (log.isDebugEnabled()) {
+ log.debug("Persisting deletes of ledgers {}", deletedLedgers);
+ }
+
+ ledgersDb.sync();
+ key.recycle();
+ }
+
+ private static final Logger log = LoggerFactory.getLogger(LedgerMetadataIndex.class);
+}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LongPairWrapper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LongPairWrapper.java
new file mode 100644
index 00000000000..993297c2b36
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LongPairWrapper.java
@@ -0,0 +1,69 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import io.netty.util.Recycler;
+import io.netty.util.Recycler.Handle;
+
+/**
+ * Recyclable wrapper that holds a pair of longs.
+ */
+class LongPairWrapper {
+
+ final byte[] array = new byte[16];
+
+ public void set(long first, long second) {
+ ArrayUtil.setLong(array, 0, first);
+ ArrayUtil.setLong(array, 8, second);
+ }
+
+ public long getFirst() {
+ return ArrayUtil.getLong(array, 0);
+ }
+
+ public long getSecond() {
+ return ArrayUtil.getLong(array, 8);
+ }
+
+ public static LongPairWrapper get(long first, long second) {
+ LongPairWrapper lp = RECYCLER.get();
+ ArrayUtil.setLong(lp.array, 0, first);
+ ArrayUtil.setLong(lp.array, 8, second);
+ return lp;
+ }
+
+ public void recycle() {
+ handle.recycle(this);
+ }
+
+ private static final Recycler RECYCLER = new Recycler() {
+ @Override
+ protected LongPairWrapper newObject(Handle handle) {
+ return new LongPairWrapper(handle);
+ }
+ };
+
+ private final Handle handle;
+
+ private LongPairWrapper(Handle handle) {
+ this.handle = handle;
+ }
+}
\ No newline at end of file
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LongWrapper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LongWrapper.java
new file mode 100644
index 00000000000..144b9ee75a1
--- /dev/null
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LongWrapper.java
@@ -0,0 +1,67 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import io.netty.util.Recycler;
+import io.netty.util.Recycler.Handle;
+
+/**
+ * Wrapper for a long serialized into a byte array.
+ */
+class LongWrapper {
+
+ final byte[] array = new byte[8];
+
+ public void set(long value) {
+ ArrayUtil.setLong(array, 0, value);
+ }
+
+ public long getValue() {
+ return ArrayUtil.getLong(array, 0);
+ }
+
+ public static LongWrapper get() {
+ return RECYCLER.get();
+ }
+
+ public static LongWrapper get(long value) {
+ LongWrapper lp = RECYCLER.get();
+ ArrayUtil.setLong(lp.array, 0, value);
+ return lp;
+ }
+
+ public void recycle() {
+ handle.recycle(this);
+ }
+
+ private static final Recycler RECYCLER = new Recycler() {
+ @Override
+ protected LongWrapper newObject(Handle handle) {
+ return new LongWrapper(handle);
+ }
+ };
+
+ private final Handle handle;
+
+ private LongWrapper(Handle handle) {
+ this.handle = handle;
+ }
+}
\ No newline at end of file
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageBookieTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageBookieTest.java
new file mode 100644
index 00000000000..da7f32a8d06
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageBookieTest.java
@@ -0,0 +1,52 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.junit.Test;
+
+/**
+ * Unit test for {@link DbLedgerStorageBookieTest}.
+ */
+public class DbLedgerStorageBookieTest extends BookKeeperClusterTestCase {
+
+ public DbLedgerStorageBookieTest() {
+ super(1);
+ baseConf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ baseConf.setFlushInterval(60000);
+ baseConf.setGcWaitTime(60000);
+ }
+
+ @Test
+ public void testRecoveryEmptyLedger() throws Exception {
+ LedgerHandle lh1 = bkc.createLedger(1, 1, DigestType.MAC, new byte[0]);
+
+ // Force ledger close & recovery
+ LedgerHandle lh2 = bkc.openLedger(lh1.getId(), DigestType.MAC, new byte[0]);
+
+ assertEquals(0, lh2.getLength());
+ assertEquals(-1, lh2.getLastAddConfirmed());
+ }
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java
new file mode 100644
index 00000000000..1e91c2ce831
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java
@@ -0,0 +1,423 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.google.common.collect.Lists;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.Bookie.NoEntryException;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.EntryLocation;
+import org.apache.bookkeeper.bookie.EntryLogger;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.proto.BookieProtocol;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Unit test for {@link DbLedgerStorage}.
+ */
+public class DbLedgerStorageTest {
+
+ private DbLedgerStorage storage;
+ private File tmpDir;
+
+ @Before
+ public void setup() throws Exception {
+ tmpDir = File.createTempFile("bkTest", ".dir");
+ tmpDir.delete();
+ tmpDir.mkdir();
+ File curDir = Bookie.getCurrentDirectory(tmpDir);
+ Bookie.checkDirectoryStructure(curDir);
+
+ int gcWaitTime = 1000;
+ ServerConfiguration conf = new ServerConfiguration();
+ conf.setGcWaitTime(gcWaitTime);
+ conf.setAllowLoopback(true);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { tmpDir.toString() });
+ Bookie bookie = new Bookie(conf);
+
+ storage = (DbLedgerStorage) bookie.getLedgerStorage();
+ }
+
+ @After
+ public void teardown() throws Exception {
+ storage.shutdown();
+ tmpDir.delete();
+ }
+
+ @Test
+ public void simple() throws Exception {
+ assertEquals(false, storage.ledgerExists(3));
+ try {
+ storage.isFenced(3);
+ fail("should have failed");
+ } catch (Bookie.NoLedgerException nle) {
+ // OK
+ }
+ assertEquals(false, storage.ledgerExists(3));
+ try {
+ storage.setFenced(3);
+ fail("should have failed");
+ } catch (Bookie.NoLedgerException nle) {
+ // OK
+ }
+ storage.setMasterKey(3, "key".getBytes());
+ try {
+ storage.setMasterKey(3, "other-key".getBytes());
+ fail("should have failed");
+ } catch (IOException ioe) {
+ assertTrue(ioe.getCause() instanceof BookieException.BookieIllegalOpException);
+ }
+ // setting the same key is NOOP
+ storage.setMasterKey(3, "key".getBytes());
+ assertEquals(true, storage.ledgerExists(3));
+ assertEquals(true, storage.setFenced(3));
+ assertEquals(true, storage.isFenced(3));
+ assertEquals(false, storage.setFenced(3));
+
+ storage.setMasterKey(4, "key".getBytes());
+ assertEquals(false, storage.isFenced(4));
+ assertEquals(true, storage.ledgerExists(4));
+
+ assertEquals("key", new String(storage.readMasterKey(4)));
+
+ assertEquals(Lists.newArrayList(4L, 3L), Lists.newArrayList(storage.getActiveLedgersInRange(0, 100)));
+ assertEquals(Lists.newArrayList(4L, 3L), Lists.newArrayList(storage.getActiveLedgersInRange(3, 100)));
+ assertEquals(Lists.newArrayList(3L), Lists.newArrayList(storage.getActiveLedgersInRange(0, 4)));
+
+ // Add / read entries
+ ByteBuf entry = Unpooled.buffer(1024);
+ entry.writeLong(4); // ledger id
+ entry.writeLong(1); // entry id
+ entry.writeBytes("entry-1".getBytes());
+
+ assertEquals(false, ((DbLedgerStorage) storage).isFlushRequired());
+
+ assertEquals(1, storage.addEntry(entry));
+
+ assertEquals(true, ((DbLedgerStorage) storage).isFlushRequired());
+
+ // Read from write cache
+ ByteBuf res = storage.getEntry(4, 1);
+ assertEquals(entry, res);
+
+ storage.flush();
+
+ assertEquals(false, ((DbLedgerStorage) storage).isFlushRequired());
+
+ // Read from db
+ res = storage.getEntry(4, 1);
+ assertEquals(entry, res);
+
+ try {
+ storage.getEntry(4, 2);
+ fail("Should have thrown exception");
+ } catch (NoEntryException e) {
+ // ok
+ }
+
+ ByteBuf entry2 = Unpooled.buffer(1024);
+ entry2.writeLong(4); // ledger id
+ entry2.writeLong(2); // entry id
+ entry2.writeBytes("entry-2".getBytes());
+
+ storage.addEntry(entry2);
+
+ // Read last entry in ledger
+ res = storage.getEntry(4, BookieProtocol.LAST_ADD_CONFIRMED);
+ assertEquals(entry2, res);
+
+ ByteBuf entry3 = Unpooled.buffer(1024);
+ entry3.writeLong(4); // ledger id
+ entry3.writeLong(3); // entry id
+ entry3.writeBytes("entry-3".getBytes());
+ storage.addEntry(entry3);
+
+ ByteBuf entry4 = Unpooled.buffer(1024);
+ entry4.writeLong(4); // ledger id
+ entry4.writeLong(4); // entry id
+ entry4.writeBytes("entry-4".getBytes());
+ storage.addEntry(entry4);
+
+ res = storage.getEntry(4, 4);
+ assertEquals(entry4, res);
+
+ // Delete
+ assertEquals(true, storage.ledgerExists(4));
+ storage.deleteLedger(4);
+ assertEquals(false, storage.ledgerExists(4));
+
+ // Should not throw exception event if the ledger was deleted
+ storage.getEntry(4, 4);
+
+ storage.addEntry(Unpooled.wrappedBuffer(entry2));
+ res = storage.getEntry(4, BookieProtocol.LAST_ADD_CONFIRMED);
+ assertEquals(entry4, res);
+
+ // Get last entry from storage
+ storage.flush();
+
+ try {
+ storage.getEntry(4, 4);
+ fail("Should have thrown exception since the ledger was deleted");
+ } catch (NoEntryException e) {
+ // ok
+ }
+ }
+
+ @Test
+ public void testBookieCompaction() throws Exception {
+ storage.setMasterKey(4, "key".getBytes());
+
+ ByteBuf entry3 = Unpooled.buffer(1024);
+ entry3.writeLong(4); // ledger id
+ entry3.writeLong(3); // entry id
+ entry3.writeBytes("entry-3".getBytes());
+ storage.addEntry(entry3);
+
+ // Simulate bookie compaction
+ EntryLogger entryLogger = ((DbLedgerStorage) storage).getEntryLogger();
+ // Rewrite entry-3
+ ByteBuf newEntry3 = Unpooled.buffer(1024);
+ newEntry3.writeLong(4); // ledger id
+ newEntry3.writeLong(3); // entry id
+ newEntry3.writeBytes("new-entry-3".getBytes());
+ long location = entryLogger.addEntry(4, newEntry3, false);
+
+ List locations = Lists.newArrayList(new EntryLocation(4, 3, location));
+ storage.updateEntriesLocations(locations);
+
+ ByteBuf res = storage.getEntry(4, 3);
+ System.out.println("res: " + ByteBufUtil.hexDump(res));
+ System.out.println("newEntry3: " + ByteBufUtil.hexDump(newEntry3));
+ assertEquals(newEntry3, res);
+ }
+
+ @Test
+ public void doubleDirectoryError() throws Exception {
+ int gcWaitTime = 1000;
+ ServerConfiguration conf = new ServerConfiguration();
+ conf.setGcWaitTime(gcWaitTime);
+ conf.setAllowLoopback(true);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { "dir1", "dir2" });
+
+ try {
+ new Bookie(conf);
+ fail("Should have failed because of the 2 directories");
+ } catch (IllegalArgumentException e) {
+ // ok
+ }
+
+ }
+
+ @Test
+ public void testRewritingEntries() throws Exception {
+ storage.setMasterKey(1, "key".getBytes());
+
+ try {
+ storage.getEntry(1, -1);
+ fail("Should throw exception");
+ } catch (Bookie.NoEntryException e) {
+ // ok
+ }
+
+ ByteBuf entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(1); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+
+ storage.addEntry(entry1);
+ storage.flush();
+
+ ByteBuf newEntry1 = Unpooled.buffer(1024);
+ newEntry1.writeLong(1); // ledger id
+ newEntry1.writeLong(1); // entry id
+ newEntry1.writeBytes("new-entry-1".getBytes());
+
+ storage.addEntry(newEntry1);
+ storage.flush();
+
+ ByteBuf response = storage.getEntry(1, 1);
+ assertEquals(newEntry1, response);
+ }
+
+ @Test
+ public void testEntriesOutOfOrder() throws Exception {
+ storage.setMasterKey(1, "key".getBytes());
+
+ ByteBuf entry2 = Unpooled.buffer(1024);
+ entry2.writeLong(1); // ledger id
+ entry2.writeLong(2); // entry id
+ entry2.writeBytes("entry-2".getBytes());
+
+ storage.addEntry(entry2);
+
+ try {
+ storage.getEntry(1, 1);
+ fail("Entry doesn't exist");
+ } catch (NoEntryException e) {
+ // Ok, entry doesn't exist
+ }
+
+ ByteBuf res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+
+ ByteBuf entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(1); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+
+ storage.addEntry(entry1);
+
+ res = storage.getEntry(1, 1);
+ assertEquals(entry1, res);
+
+ res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+
+ storage.flush();
+
+ res = storage.getEntry(1, 1);
+ assertEquals(entry1, res);
+
+ res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+ }
+
+ @Test
+ public void testEntriesOutOfOrderWithFlush() throws Exception {
+ storage.setMasterKey(1, "key".getBytes());
+
+ ByteBuf entry2 = Unpooled.buffer(1024);
+ entry2.writeLong(1); // ledger id
+ entry2.writeLong(2); // entry id
+ entry2.writeBytes("entry-2".getBytes());
+
+ storage.addEntry(entry2);
+
+ try {
+ storage.getEntry(1, 1);
+ fail("Entry doesn't exist");
+ } catch (NoEntryException e) {
+ // Ok, entry doesn't exist
+ }
+
+ ByteBuf res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+ res.release();
+
+ storage.flush();
+
+ try {
+ storage.getEntry(1, 1);
+ fail("Entry doesn't exist");
+ } catch (NoEntryException e) {
+ // Ok, entry doesn't exist
+ }
+
+ res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+ res.release();
+
+ ByteBuf entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(1); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+
+ storage.addEntry(entry1);
+
+ res = storage.getEntry(1, 1);
+ assertEquals(entry1, res);
+ res.release();
+
+ res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+ res.release();
+
+ storage.flush();
+
+ res = storage.getEntry(1, 1);
+ assertEquals(entry1, res);
+ res.release();
+
+ res = storage.getEntry(1, 2);
+ assertEquals(entry2, res);
+ res.release();
+ }
+
+ @Test
+ public void testAddEntriesAfterDelete() throws Exception {
+ storage.setMasterKey(1, "key".getBytes());
+
+ ByteBuf entry0 = Unpooled.buffer(1024);
+ entry0.writeLong(1); // ledger id
+ entry0.writeLong(0); // entry id
+ entry0.writeBytes("entry-0".getBytes());
+
+ ByteBuf entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(1); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+
+ storage.addEntry(entry0);
+ storage.addEntry(entry1);
+
+ storage.flush();
+
+ storage.deleteLedger(1);
+
+ storage.setMasterKey(1, "key".getBytes());
+
+ entry0 = Unpooled.buffer(1024);
+ entry0.writeLong(1); // ledger id
+ entry0.writeLong(0); // entry id
+ entry0.writeBytes("entry-0".getBytes());
+
+ entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(1); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+
+ storage.addEntry(entry0);
+ storage.addEntry(entry1);
+
+ assertEquals(entry0, storage.getEntry(1, 0));
+ assertEquals(entry1, storage.getEntry(1, 1));
+
+ storage.flush();
+ }
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java
new file mode 100644
index 00000000000..0d3f5bbc86e
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageWriteCacheTest.java
@@ -0,0 +1,137 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Unit test for {@link DbLedgerStorage}.
+ */
+public class DbLedgerStorageWriteCacheTest {
+
+ private DbLedgerStorage storage;
+ private File tmpDir;
+
+ private static class MockedDbLedgerStorage extends DbLedgerStorage {
+
+ @Override
+ public void flush() throws IOException {
+ flushMutex.lock();
+ try {
+ // Swap the write caches and block indefinitely to simulate a slow disk
+ WriteCache tmp = writeCacheBeingFlushed;
+ writeCacheBeingFlushed = writeCache;
+ writeCache = tmp;
+
+ // since the cache is switched, we can allow flush to be triggered
+ hasFlushBeenTriggered.set(false);
+
+ // Block the flushing thread
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ return;
+ }
+ } finally {
+ flushMutex.unlock();
+ }
+ }
+
+ }
+
+ @Before
+ public void setup() throws Exception {
+ tmpDir = File.createTempFile("bkTest", ".dir");
+ tmpDir.delete();
+ tmpDir.mkdir();
+ File curDir = Bookie.getCurrentDirectory(tmpDir);
+ Bookie.checkDirectoryStructure(curDir);
+
+ int gcWaitTime = 1000;
+ ServerConfiguration conf = new ServerConfiguration();
+ conf.setGcWaitTime(gcWaitTime);
+ conf.setAllowLoopback(true);
+ conf.setLedgerStorageClass(MockedDbLedgerStorage.class.getName());
+ conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 1);
+ conf.setLedgerDirNames(new String[] { tmpDir.toString() });
+ Bookie bookie = new Bookie(conf);
+
+ storage = (DbLedgerStorage) bookie.getLedgerStorage();
+ }
+
+ @After
+ public void teardown() throws Exception {
+ storage.shutdown();
+ tmpDir.delete();
+ }
+
+ @Test
+ public void writeCacheFull() throws Exception {
+ storage.setMasterKey(4, "key".getBytes());
+ assertEquals(false, storage.isFenced(4));
+ assertEquals(true, storage.ledgerExists(4));
+
+ assertEquals("key", new String(storage.readMasterKey(4)));
+
+ // Add enough entries to fill the 1st write cache
+ for (int i = 0; i < 5; i++) {
+ ByteBuf entry = Unpooled.buffer(100 * 1024 + 2 * 8);
+ entry.writeLong(4); // ledger id
+ entry.writeLong(i); // entry id
+ entry.writeZero(100 * 1024);
+ storage.addEntry(entry);
+ }
+
+ for (int i = 0; i < 5; i++) {
+ ByteBuf entry = Unpooled.buffer(100 * 1024 + 2 * 8);
+ entry.writeLong(4); // ledger id
+ entry.writeLong(5 + i); // entry id
+ entry.writeZero(100 * 1024);
+ storage.addEntry(entry);
+ }
+
+ // Next add should fail for cache full
+ ByteBuf entry = Unpooled.buffer(100 * 1024 + 2 * 8);
+ entry.writeLong(4); // ledger id
+ entry.writeLong(22); // entry id
+ entry.writeZero(100 * 1024);
+
+ try {
+ storage.addEntry(entry);
+ fail("Should have thrown exception");
+ } catch (IOException e) {
+ // Expected
+ }
+ }
+}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndexTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndexTest.java
new file mode 100644
index 00000000000..6e83ffd58a4
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndexTest.java
@@ -0,0 +1,111 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.bookie.storage.ldb;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.junit.Test;
+
+/**
+ * Unit test for {@link EntryLocationIndex}.
+ */
+public class EntryLocationIndexTest {
+
+ private final ServerConfiguration serverConfiguration = new ServerConfiguration();
+
+ @Test
+ public void deleteLedgerTest() 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);
+
+ // Add some dummy indexes
+ idx.addLocation(40312, 0, 1);
+ idx.addLocation(40313, 10, 2);
+ idx.addLocation(40320, 0, 3);
+
+ // Add more indexes in a different batch
+ idx.addLocation(40313, 11, 5);
+ idx.addLocation(40313, 12, 6);
+ idx.addLocation(40320, 1, 7);
+ idx.addLocation(40312, 3, 4);
+
+ idx.delete(40313);
+
+ assertEquals(1, idx.getLocation(40312, 0));
+ assertEquals(4, idx.getLocation(40312, 3));
+ assertEquals(3, idx.getLocation(40320, 0));
+ assertEquals(7, idx.getLocation(40320, 1));
+
+ assertEquals(2, idx.getLocation(40313, 10));
+ assertEquals(5, idx.getLocation(40313, 11));
+ assertEquals(6, idx.getLocation(40313, 12));
+
+ idx.removeOffsetFromDeletedLedgers();
+
+ // After flush the keys will be removed
+ assertEquals(0, idx.getLocation(40313, 10));
+ assertEquals(0, idx.getLocation(40313, 11));
+ assertEquals(0, idx.getLocation(40313, 12));
+
+ idx.close();
+ }
+
+ // this tests if a ledger is added after it has been deleted
+ @Test
+ public void addLedgerAfterDeleteTest() 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);
+
+ // Add some dummy indexes
+ idx.addLocation(40312, 0, 1);
+ idx.addLocation(40313, 10, 2);
+ idx.addLocation(40320, 0, 3);
+
+ idx.delete(40313);
+
+ // Add more indexes in a different batch
+ idx.addLocation(40313, 11, 5);
+ idx.addLocation(40313, 12, 6);
+ idx.addLocation(40320, 1, 7);
+ idx.addLocation(40312, 3, 4);
+
+ idx.removeOffsetFromDeletedLedgers();
+
+ assertEquals(0, idx.getLocation(40313, 11));
+ assertEquals(0, idx.getLocation(40313, 12));
+
+ idx.close();
+ }
+}
diff --git a/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml b/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
index 2cc5ce766cd..736c88b11d4 100644
--- a/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
+++ b/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
@@ -24,6 +24,10 @@
+
+
+
+