Fix skipped checkpoint flushes that can cause data loss in multi-journal DbLedgerStorage

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/CheckpointSource.java

@@ -29,6 +29,23 @@ public interface CheckpointSource {
      */
     interface Checkpoint extends Comparable<Checkpoint> {
 
+        /**
+         * Return whether this checkpoint covers all state represented by another
+         * checkpoint and additionally represents later state.
+         *
+         * <p>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

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)

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;
         }
 

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);
+        }
+    }
+}

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);
+        }
+    }
 }