diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSource.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSource.java index 2221632469a..bca2ed72302 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSource.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSource.java @@ -29,6 +29,23 @@ public interface CheckpointSource { */ interface Checkpoint extends Comparable { + /** + * Return whether this checkpoint covers all state represented by another + * checkpoint and additionally represents later state. + * + *

For checkpoints that aggregate multiple independent sources, this means + * every component covers the corresponding component in {@code other} and at + * least one component is strictly later. Such implementations may override + * this method to define that partial ordering without changing the total + * ordering provided by {@link #compareTo(Checkpoint)}. + * + * @param other the checkpoint to compare against + * @return true when this checkpoint is strictly after {@code other} + */ + default boolean isAfter(Checkpoint other) { + return compareTo(other) > 0; + } + Checkpoint MAX = new Checkpoint() { @Override diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSourceList.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSourceList.java index 0c156abc965..bed6abfe8c4 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSourceList.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSourceList.java @@ -110,6 +110,35 @@ public int compareTo(Checkpoint o) { return 0; } + @Override + public boolean isAfter(Checkpoint o) { + if (o == Checkpoint.MAX) { + return false; + } else if (o == Checkpoint.MIN) { + return true; + } + + checkArgument(o instanceof CheckpointList); + CheckpointList other = (CheckpointList) o; + if (checkpoints.size() != other.checkpoints.size()) { + return false; + } + + boolean strictlyAfter = false; + for (int i = 0; i < checkpoints.size(); i++) { + Checkpoint checkpoint = checkpoints.get(i); + Checkpoint otherCheckpoint = other.checkpoints.get(i); + if (checkpoint.compareTo(otherCheckpoint) == 0) { + continue; + } + if (!checkpoint.isAfter(otherCheckpoint)) { + return false; + } + strictlyAfter = true; + } + return strictlyAfter; + } + @Override public String toString() { return MoreObjects.toStringHelper(CheckpointList.class) diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java index 6778de2f824..81f87281e11 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java @@ -790,7 +790,7 @@ boolean isFlushRequired() { @Override public void checkpoint(Checkpoint checkpoint) throws IOException { Checkpoint thisCheckpoint = checkpointSource.newCheckpoint(); - if (lastCheckpoint.compareTo(checkpoint) > 0) { + if (lastCheckpoint.isAfter(checkpoint)) { return; } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CheckpointSourceListTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CheckpointSourceListTest.java new file mode 100644 index 00000000000..682e7c0905b --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CheckpointSourceListTest.java @@ -0,0 +1,131 @@ +/* + * + * 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; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.google.common.collect.Lists; +import java.io.IOException; +import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint; +import org.junit.Test; + +/** + * Tests for {@link CheckpointSourceList}. + */ +public class CheckpointSourceListTest { + + private final TestCheckpointSource firstSource = new TestCheckpointSource(); + private final TestCheckpointSource secondSource = new TestCheckpointSource(); + private final CheckpointSourceList checkpointSource = + new CheckpointSourceList(Lists.newArrayList(firstSource, secondSource)); + + @Test + public void testIncomparableCheckpointsAreNotAfterEachOther() { + Checkpoint first = newCheckpoint(2, 1); + Checkpoint second = newCheckpoint(1, 2); + + assertTrue(first.compareTo(second) > 0); + assertFalse(first.isAfter(second)); + assertFalse(second.isAfter(first)); + } + + @Test + public void testCheckpointIsAfterOnlyWhenAllComponentsAreCovered() { + Checkpoint requested = newCheckpoint(1, 2); + Checkpoint after = newCheckpoint(2, 2); + Checkpoint equal = newCheckpoint(1, 2); + Checkpoint before = newCheckpoint(1, 1); + + assertTrue(after.isAfter(requested)); + assertFalse(equal.isAfter(requested)); + assertFalse(before.isAfter(requested)); + } + + @Test + public void testMinAndMaxCheckpoints() { + Checkpoint checkpoint = newCheckpoint(1, 1); + + assertTrue(checkpoint.isAfter(Checkpoint.MIN)); + assertFalse(checkpoint.isAfter(Checkpoint.MAX)); + assertTrue(Checkpoint.MAX.isAfter(checkpoint)); + assertFalse(Checkpoint.MIN.isAfter(checkpoint)); + assertFalse(Checkpoint.MAX.isAfter(Checkpoint.MAX)); + assertFalse(Checkpoint.MIN.isAfter(Checkpoint.MIN)); + } + + @Test + public void testSingleSourceKeepsScalarOrdering() { + TestCheckpointSource source = new TestCheckpointSource(); + CheckpointSourceList singleSource = new CheckpointSourceList(Lists.newArrayList(source)); + + source.setValue(1); + Checkpoint first = singleSource.newCheckpoint(); + source.setValue(2); + Checkpoint second = singleSource.newCheckpoint(); + + assertTrue(second.isAfter(first)); + assertFalse(first.isAfter(second)); + assertFalse(first.isAfter(first)); + } + + private Checkpoint newCheckpoint(long first, long second) { + firstSource.setValue(first); + secondSource.setValue(second); + return checkpointSource.newCheckpoint(); + } + + private static final class TestCheckpointSource implements CheckpointSource { + private long value; + + void setValue(long value) { + this.value = value; + } + + @Override + public Checkpoint newCheckpoint() { + return new TestCheckpoint(value); + } + + @Override + public void checkpointComplete(Checkpoint checkpoint, boolean compact) throws IOException { + // No-op + } + } + + private static final class TestCheckpoint implements Checkpoint { + private final long value; + + TestCheckpoint(long value) { + this.value = value; + } + + @Override + public int compareTo(Checkpoint other) { + if (other == Checkpoint.MAX) { + return -1; + } else if (other == Checkpoint.MIN) { + return 1; + } + return Long.compare(value, ((TestCheckpoint) other).value); + } + } +} 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 index a569a638ccf..3736b284d83 100644 --- 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 @@ -48,6 +48,7 @@ import org.apache.bookkeeper.bookie.BookieException; import org.apache.bookkeeper.bookie.BookieImpl; import org.apache.bookkeeper.bookie.CheckpointSource; +import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint; import org.apache.bookkeeper.bookie.CheckpointSourceList; import org.apache.bookkeeper.bookie.DefaultEntryLogger; import org.apache.bookkeeper.bookie.EntryLocation; @@ -232,6 +233,51 @@ public void simple() throws Exception { } } + @Test + public void testCheckpointDoesNotSkipIncomparableCompositeCheckpoint() throws Exception { + TestCheckpointSource firstSource = new TestCheckpointSource(); + TestCheckpointSource secondSource = new TestCheckpointSource(); + CheckpointSourceList checkpointSource = + new CheckpointSourceList(Lists.newArrayList(firstSource, secondSource)); + SingleDirectoryDbLedgerStorage singleDirectoryStorage = storage.getLedgerStorageList().get(0); + singleDirectoryStorage.setCheckpointSource(checkpointSource); + + // Model two aggregate snapshots sampled concurrently: one observes a newer + // first source while the other observes a newer second source. + firstSource.setValue(2); + secondSource.setValue(1); + CheckpointSource.Checkpoint firstCheckpoint = checkpointSource.newCheckpoint(); + + ByteBuf firstEntry = Unpooled.buffer(24); + firstEntry.writeLong(1); + firstEntry.writeLong(1); + firstEntry.writeLong(0); + + ByteBuf secondEntry = Unpooled.buffer(24); + secondEntry.writeLong(1); + secondEntry.writeLong(2); + secondEntry.writeLong(1); + + try { + storage.addEntry(firstEntry); + singleDirectoryStorage.checkpoint(firstCheckpoint); + assertFalse(singleDirectoryStorage.isFlushRequired()); + + firstSource.setValue(1); + secondSource.setValue(2); + CheckpointSource.Checkpoint requestedCheckpoint = checkpointSource.newCheckpoint(); + + storage.addEntry(secondEntry); + assertTrue(singleDirectoryStorage.isFlushRequired()); + + singleDirectoryStorage.checkpoint(requestedCheckpoint); + assertFalse(singleDirectoryStorage.isFlushRequired()); + } finally { + ReferenceCountUtil.safeRelease(firstEntry); + ReferenceCountUtil.safeRelease(secondEntry); + } + } + @Test public void testBookieCompaction() throws Exception { storage.setMasterKey(4, "key".getBytes()); @@ -1121,4 +1167,40 @@ protected void persistLastLogMark(LastLogMark mark) throws NoWritableLedgerDirEx } } } + + private static final class TestCheckpointSource implements CheckpointSource { + private long value; + + void setValue(long value) { + this.value = value; + } + + @Override + public Checkpoint newCheckpoint() { + return new TestCheckpoint(value); + } + + @Override + public void checkpointComplete(Checkpoint checkpoint, boolean compact) throws IOException { + // No-op + } + } + + private static final class TestCheckpoint implements Checkpoint { + private final long value; + + TestCheckpoint(long value) { + this.value = value; + } + + @Override + public int compareTo(Checkpoint other) { + if (other == Checkpoint.MAX) { + return -1; + } else if (other == Checkpoint.MIN) { + return 1; + } + return Long.compare(value, ((TestCheckpoint) other).value); + } + } }