BP-46: Data integrity check for running without journal

bookkeeper-common/build.gradle

@@ -38,6 +38,7 @@ dependencies {
     implementation depLibs.jacksonDatabind
     implementation depLibs.jctools
     implementation depLibs.nettyCommon
+    implementation depLibs.rxjava
     implementation depLibs.slf4j
     testCompileOnly depLibs.lombok
     testImplementation depLibs.commonsLang3

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/component/RxSchedulerLifecycleComponent.java

@@ -0,0 +1,60 @@
+/*
+ * 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.common.component;
+
+import io.reactivex.rxjava3.core.Scheduler;
+import java.util.concurrent.ExecutorService;
+import org.apache.bookkeeper.common.conf.ComponentConfiguration;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * The scheduler for rxjava based jobs, such as data integrity checking.
+ */
+public class RxSchedulerLifecycleComponent extends AbstractLifecycleComponent<ComponentConfiguration> {
+    private final Scheduler scheduler;
+    private final ExecutorService rxExecutor;
+
+    public RxSchedulerLifecycleComponent(String componentName,
+                                         ComponentConfiguration conf,
+                                         StatsLogger stats,
+                                         Scheduler scheduler,
+                                         ExecutorService rxExecutor) {
+        super(componentName, conf, stats);
+        this.scheduler = scheduler;
+        this.rxExecutor = rxExecutor;
+    }
+
+    @Override
+    protected void doStart() {
+        scheduler.start();
+    }
+
+    @Override
+    protected void doStop() {
+        scheduler.shutdown();
+        rxExecutor.shutdown();
+    }
+
+    @Override
+    public void doClose() {
+        scheduler.shutdown();
+        rxExecutor.shutdown();
+    }
+}

bookkeeper-common/src/test/java/org/apache/bookkeeper/common/util/MockTicker.java

@@ -0,0 +1,42 @@
+/*
+ * 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.common.util;
+
+import com.google.common.base.Ticker;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Test implementation of Ticker.
+ */
+public class MockTicker extends Ticker {
+    private AtomicLong tick = new AtomicLong(0);
+
+    public void advance(int period, TimeUnit unit) {
+        tick.addAndGet(unit.toNanos(period));
+    }
+
+    @Override
+    public long read() {
+        return tick.get();
+    }
+}
+
+

bookkeeper-dist/all/build.gradle

@@ -64,6 +64,7 @@ def depLicences = [
         "paranamer-2.8/LICENSE.txt",
         "protobuf-3.14.0/LICENSE",
         "protobuf-3.12.0/LICENSE",
+        "reactivestreams-1.0.3/LICENSE",
         "scala-library-2.11.7/LICENSE.md",
         "scala-parser-combinators_2.11-1.0.4/LICENSE.md",
         "scala-reflect-2.11.8/LICENSE.md",

bookkeeper-dist/bkctl/build.gradle

@@ -53,6 +53,7 @@ def depLicences = [
         "bouncycastle-1.0.2.1/LICENSE.html",
         "protobuf-3.14.0/LICENSE",
         "protobuf-3.12.0/LICENSE",
+        "reactivestreams-1.0.3/LICENSE",
         "slf4j-1.7.32/LICENSE.txt",
 ]
 

bookkeeper-dist/server/build.gradle

@@ -57,6 +57,7 @@ def depLicences = [
         "bouncycastle-1.0.2.1/LICENSE.html",
         "protobuf-3.14.0/LICENSE",
         "protobuf-3.12.0/LICENSE",
+        "reactivestreams-1.0.3/LICENSE",
         "slf4j-1.7.32/LICENSE.txt",
 ]
 distributions {

bookkeeper-dist/src/assemble/bin-all.xml

@@ -63,6 +63,7 @@
         <include>paranamer-2.8/LICENSE.txt</include>
         <include>protobuf-3.14.0/LICENSE</include>
         <include>protobuf-3.12.0/LICENSE</include>
+        <include>reactivestreams-1.0.3/LICENSE</include>
         <include>scala-library-2.11.7/LICENSE.md</include>
         <include>scala-parser-combinators_2.11-1.0.4/LICENSE.md</include>
         <include>scala-reflect-2.11.8/LICENSE.md</include>

bookkeeper-dist/src/assemble/bin-server.xml

@@ -56,6 +56,7 @@
         <include>bouncycastle-1.0.2.1/LICENSE.html</include>
         <include>protobuf-3.14.0/LICENSE</include>
         <include>protobuf-3.12.0/LICENSE</include>
+        <include>reactivestreams-1.0.3/LICENSE</include>
         <include>slf4j-1.7.32/LICENSE.txt</include>
       </includes>
       <fileMode>644</fileMode>

bookkeeper-dist/src/assemble/bkctl.xml

@@ -70,6 +70,7 @@
         <include>bouncycastle-1.0.2.1/LICENSE.html</include>
         <include>protobuf-3.14.0/LICENSE</include>
         <include>protobuf-3.12.0/LICENSE</include>
+        <include>reactivestreams-1.0.3/LICENSE</include>
         <include>slf4j-1.7.32/LICENSE.txt</include>
       </includes>
       <fileMode>644</fileMode>

bookkeeper-dist/src/main/resources/LICENSE-all.bin.txt

@@ -307,6 +307,7 @@ Apache Software License, Version 2.
 - lib/io.perfmark-perfmark-api-0.23.0.jar [48]
 - lib/org.conscrypt-conscrypt-openjdk-uber-2.5.1.jar [49]
 - lib/org.xerial.snappy-snappy-java-1.1.7.jar [50]
+- lib/io.reactivex.rxjava3-rxjava-3.0.1.jar [51]
 
 [1] Source available at https://github.com/FasterXML/jackson-annotations/tree/jackson-annotations-2.11.0
 [2] Source available at https://github.com/FasterXML/jackson-core/tree/jackson-core-2.11.3
@@ -355,6 +356,7 @@ Apache Software License, Version 2.
 [48] Source available at https://github.com/perfmark/perfmark/releases/tag/v0.23.0
 [49] Source available at https://github.com/google/conscrypt/releases/tag/2.5.1
 [50] Source available at https://github.com/google/snappy/releases/tag/1.1.7
+[51] Source available at https://github.com/ReactiveX/RxJava/tree/v3.0.1
 
 ------------------------------------------------------------------------------------
 lib/io.netty-netty-codec-4.1.72.Final.jar bundles some 3rd party dependencies
@@ -666,3 +668,9 @@ MIT License. For details, see deps/checker-qual-3.5.0/LICENSE
 
 Bundles as
   - lib/org.checkerframework-checker-qual-3.5.0.jar
+------------------------------------------------------------------------------------
+This product bundles the Reactive Streams library, which is licensed under
+Public Domain (CC0). For details, see deps/reactivestreams-1.0.3/LICENSE
+
+Bundles as
+  - lib/org.reactivestreams-reactive-streams-1.0.3.jar

bookkeeper-dist/src/main/resources/LICENSE-bkctl.bin.txt

@@ -284,6 +284,7 @@ Apache Software License, Version 2.
 - lib/io.perfmark-perfmark-api-0.23.0.jar [47]
 - lib/org.conscrypt-conscrypt-openjdk-uber-2.5.1.jar [49]
 - lib/org.xerial.snappy-snappy-java-1.1.7.jar [50]
+- lib/io.reactivex.rxjava3-rxjava-3.0.1.jar [51]
 
 [1] Source available at https://github.com/FasterXML/jackson-annotations/tree/jackson-annotations-2.11.0
 [2] Source available at https://github.com/FasterXML/jackson-core/tree/jackson-core-2.11.3
@@ -324,6 +325,7 @@ Apache Software License, Version 2.
 [47] Source available at https://github.com/perfmark/perfmark/releases/tag/v0.23.0
 [49] Source available at https://github.com/google/conscrypt/releases/tag/2.5.1
 [50] Source available at https://github.com/google/snappy/releases/tag/1.1.7
+[51] Source available at https://github.com/ReactiveX/RxJava/tree/v3.0.1
 
 ------------------------------------------------------------------------------------
 lib/io.netty-netty-codec-4.1.72.Final.jar bundles some 3rd party dependencies
@@ -593,3 +595,9 @@ MIT License. For details, see deps/checker-qual-3.5.0/LICENSE
 
 Bundles as
   - lib/org.checkerframework-checker-qual-3.5.0.jar
+------------------------------------------------------------------------------------
+This product bundles the Reactive Streams library, which is licensed under
+Public Domain (CC0). For details, see deps/reactivestreams-1.0.3/LICENSE
+
+Bundles as
+  - lib/org.reactivestreams-reactive-streams-1.0.3.jar

bookkeeper-dist/src/main/resources/LICENSE-server.bin.txt

@@ -305,6 +305,7 @@ Apache Software License, Version 2.
 - lib/io.perfmark-perfmark-api-0.23.0.jar [48]
 - lib/org.conscrypt-conscrypt-openjdk-uber-2.5.1.jar [49]
 - lib/org.xerial.snappy-snappy-java-1.1.7.jar [50]
+- lib/io.reactivex.rxjava3-rxjava-3.0.1.jar [51]
 
 [1] Source available at https://github.com/FasterXML/jackson-annotations/tree/jackson-annotations-2.11.0
 [2] Source available at https://github.com/FasterXML/jackson-core/tree/jackson-core-2.11.3
@@ -353,6 +354,7 @@ Apache Software License, Version 2.
 [48] Source available at https://github.com/perfmark/perfmark/releases/tag/v0.23.0
 [49] Source available at https://github.com/google/conscrypt/releases/tag/2.5.1
 [50] Source available at https://github.com/google/snappy/releases/tag/1.1.7
+[51] Source available at https://github.com/ReactiveX/RxJava/tree/v3.0.1
 
 ------------------------------------------------------------------------------------
 lib/io.netty-netty-codec-4.1.72.Final.jar bundles some 3rd party dependencies
@@ -658,3 +660,9 @@ MIT License. For details, see deps/checker-qual-3.5.0/LICENSE
 
 Bundles as
   - lib/org.checkerframework-checker-qual-3.5.0.jar
+------------------------------------------------------------------------------------
+This product bundles the Reactive Streams library, which is licensed under
+Public Domain (CC0). For details, see deps/reactivestreams-1.0.3/LICENSE
+
+Bundles as
+  - lib/org.reactivestreams-reactive-streams-1.0.3.jar

bookkeeper-dist/src/main/resources/deps/reactivestreams-1.0.3/LICENSE

@@ -0,0 +1,8 @@
+Licensed under Public Domain (CC0)
+
+To the extent possible under law, the person who associated CC0 with
+this code has waived all copyright and related or neighboring
+rights to this code.
+
+You should have received a copy of the CC0 legalcode along with this
+work. If not, see <http://creativecommons.org/publicdomain/zero/1.0/>.

bookkeeper-proto/src/main/proto/BookkeeperProtocol.proto

@@ -46,6 +46,7 @@ enum StatusCode {
     EFENCED = 504;
     EREADONLY = 505;
     ETOOMANYREQUESTS = 506;
+    EUNKNOWNLEDGERSTATE = 507;
 }
 
 /**

bookkeeper-proto/src/main/proto/DbLedgerStorageDataFormats.proto

@@ -28,4 +28,5 @@ message LedgerData {
     required bool fenced = 2;
     required bytes masterKey = 3;
     optional bytes explicitLac = 4;
+    optional bool limbo = 5;
 }

bookkeeper-server/build.gradle

@@ -52,6 +52,7 @@ dependencies {
     implementation depLibs.nettyTransportNativeEpoll
     implementation depLibs.protobuf
     implementation depLibs.rocksDb
+    implementation depLibs.rxjava
     implementation depLibs.zookeeper
     runtimeOnly depLibs.metricsCore
     runtimeOnly depLibs.snappy

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

@@ -48,16 +48,16 @@ void recoveryAddEntry(ByteBuf entry, WriteCallback cb, Object ctx, byte[] master
     void forceLedger(long ledgerId, WriteCallback cb, Object ctx);
     void setExplicitLac(ByteBuf entry, WriteCallback writeCallback, Object ctx, byte[] masterKey)
             throws IOException, InterruptedException, BookieException;
-    ByteBuf getExplicitLac(long ledgerId) throws IOException, NoLedgerException;
+    ByteBuf getExplicitLac(long ledgerId) throws IOException, NoLedgerException, BookieException;
 
     // these can probably be moved out and called directly on ledgerdirmanager
     long getTotalDiskSpace() throws IOException;
     long getTotalFreeSpace() throws IOException;
 
     // TODO: Shouldn't this be async?
     ByteBuf readEntry(long ledgerId, long entryId)
-            throws IOException, NoLedgerException;
-    long readLastAddConfirmed(long ledgerId) throws IOException;
+            throws IOException, NoLedgerException, BookieException;
+    long readLastAddConfirmed(long ledgerId) throws IOException, BookieException;
     PrimitiveIterator.OfLong getListOfEntriesOfLedger(long ledgerId) throws IOException, NoLedgerException;
 
     /**

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

@@ -69,6 +69,8 @@ public static BookieException create(int code) {
             return new MetadataStoreException();
         case Code.UnknownBookieIdException:
             return new UnknownBookieIdException();
+        case Code.DataUnknownException:
+            return new DataUnknownException();
         default:
             return new BookieIllegalOpException();
         }
@@ -92,6 +94,7 @@ public interface Code {
         int OperationRejectedException = -108;
         int CookieExistsException = -109;
         int EntryLogMetadataMapException = -110;
+        int DataUnknownException = -111;
     }
 
     public int getCode() {
@@ -137,6 +140,9 @@ public String getMessage(int code) {
         case Code.OperationRejectedException:
             err = "Operation rejected";
             break;
+        case Code.DataUnknownException:
+            err = "Unable to respond, ledger is in unknown state";
+            break;
         default:
             err = "Invalid operation";
             break;
@@ -337,4 +343,25 @@ public UnknownBookieIdException(Throwable cause) {
         }
     }
 
+    /**
+     * Signal when a ledger is in a limbo state and certain operations
+     * cannot be performed on it.
+     */
+    public static class DataUnknownException extends BookieException {
+        public DataUnknownException() {
+            super(Code.DataUnknownException);
+        }
+
+        public DataUnknownException(Throwable t) {
+            super(Code.DataUnknownException, t);
+        }
+
+        public DataUnknownException(String reason) {
+            super(Code.DataUnknownException, reason);
+        }
+
+        public DataUnknownException(String reason, Throwable t) {
+            super(Code.DataUnknownException, reason, t);
+        }
+    }
 }