direct-io: add support for bypassing operating system I/O cache when logging entries

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/ExceptionMessageHelper.java

@@ -0,0 +1,49 @@
+/**
+ * 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;
+
+/**
+ * Utility to make it easier to add context to exception messages.
+ */
+public class ExceptionMessageHelper {
+    public StringBuilder sb = new StringBuilder();
+    private boolean firstKV = true;
+
+    public static ExceptionMessageHelper exMsg(String msg) {
+        return new ExceptionMessageHelper(msg);
+    }
+
+    ExceptionMessageHelper(String msg) {
+        sb.append(msg).append("(");
+    }
+
+    public ExceptionMessageHelper kv(String key, Object value) {
+        if (firstKV) {
+            firstKV = false;
+        } else {
+            sb.append(",");
+        }
+        sb.append(key).append("=").append(value.toString());
+        return this;
+    }
+
+    public String toString() {
+        return sb.append(")").toString();
+    }
+}

bookkeeper-dist/all/build.gradle

@@ -68,7 +68,7 @@ def depLicences = [
         "scala-library-2.11.7/LICENSE.md",
         "scala-parser-combinators_2.11-1.0.4/LICENSE.md",
         "scala-reflect-2.11.8/LICENSE.md",
-        "slf4j-1.7.32/LICENSE.txt",
+        "slf4j-1.7.36/LICENSE.txt",
 ]
 
 distributions {

bookkeeper-dist/bkctl/build.gradle

@@ -54,7 +54,7 @@ def depLicences = [
         "protobuf-3.14.0/LICENSE",
         "protobuf-3.12.0/LICENSE",
         "reactivestreams-1.0.3/LICENSE",
-        "slf4j-1.7.32/LICENSE.txt",
+        "slf4j-1.7.36/LICENSE.txt",
 ]
 
 distributions {

bookkeeper-dist/server/build.gradle

@@ -58,7 +58,7 @@ def depLicences = [
         "protobuf-3.14.0/LICENSE",
         "protobuf-3.12.0/LICENSE",
         "reactivestreams-1.0.3/LICENSE",
-        "slf4j-1.7.32/LICENSE.txt",
+        "slf4j-1.7.36/LICENSE.txt",
 ]
 distributions {
     main {

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

@@ -67,7 +67,7 @@
         <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>
-        <include>slf4j-1.7.32/LICENSE.txt</include>
+        <include>slf4j-1.7.36/LICENSE.txt</include>
       </includes>
       <fileMode>644</fileMode>
     </fileSet>

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

@@ -57,7 +57,7 @@
         <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>
+        <include>slf4j-1.7.36/LICENSE.txt</include>
       </includes>
       <fileMode>644</fileMode>
     </fileSet>

bookkeeper-dist/src/assemble/bkctl.xml

@@ -74,7 +74,7 @@
         <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>
+        <include>slf4j-1.7.36/LICENSE.txt</include>
       </includes>
       <fileMode>644</fileMode>
     </fileSet>

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

@@ -648,11 +648,11 @@ Bundled as lib/javax.servlet-javax.servlet-api-4.0.0.jar
 Source available at https://github.com/javaee/servlet-spec/tree/4.0.0
 ------------------------------------------------------------------------------------
 This product bundles Simple Logging Facade for Java, which is available under a
-MIT license. For details, see deps/slf4j-1.7.32/LICENSE.txt.
+MIT license. For details, see deps/slf4j-1.7.36/LICENSE.txt.
 
 Bundled as
-  - lib/org.slf4j-slf4j-api-1.7.32.jar
-Source available at https://github.com/qos-ch/slf4j/tree/v_1.7.32
+  - lib/org.slf4j-slf4j-api-1.7.36.jar
+Source available at https://github.com/qos-ch/slf4j/tree/v_1.7.36
 ------------------------------------------------------------------------------------
 This product bundles the Google Auth Library, which is available under a "3-clause BSD"
 license. For details, see deps/google-auth-library-credentials-0.20.0/LICENSE

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

@@ -574,11 +574,11 @@ Source available at https://github.com/protocolbuffers/protobuf/tree/v3.19.2
 For details, see deps/protobuf-3.12.0/LICENSE.
 ------------------------------------------------------------------------------------
 This product bundles Simple Logging Facade for Java, which is available under a
-MIT license. For details, see deps/slf4j-1.7.32/LICENSE.txt.
+MIT license. For details, see deps/slf4j-1.7.36/LICENSE.txt.
 
 Bundled as
-  - lib/org.slf4j-slf4j-api-1.7.32.jar
-Source available at https://github.com/qos-ch/slf4j/tree/v_1.7.32
+  - lib/org.slf4j-slf4j-api-1.7.36.jar
+Source available at https://github.com/qos-ch/slf4j/tree/v_1.7.36
 ------------------------------------------------------------------------------------
 This product bundles the Google Auth Library, which is available under a "3-clause BSD"
 license. For details, see deps/google-auth-library-credentials-0.20.0/LICENSE

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

@@ -640,11 +640,11 @@ Bundled as lib/javax.servlet-javax.servlet-api-4.0.0.jar
 Source available at https://github.com/javaee/servlet-spec/tree/4.0.0
 ------------------------------------------------------------------------------------
 This product bundles Simple Logging Facade for Java, which is available under a
-MIT license. For details, see deps/slf4j-1.7.32/LICENSE.txt.
+MIT license. For details, see deps/slf4j-1.7.36/LICENSE.txt.
 
 Bundled as
-  - lib/org.slf4j-slf4j-api-1.7.32.jar
-Source available at https://github.com/qos-ch/slf4j/tree/v_1.7.32
+  - lib/org.slf4j-slf4j-api-1.7.36.jar
+Source available at https://github.com/qos-ch/slf4j/tree/v_1.7.36
 ------------------------------------------------------------------------------------
 This product bundles the Google Auth Library, which is available under a "3-clause BSD"
 license. For details, see deps/google-auth-library-credentials-0.20.0/LICENSE

bookkeeper-server/build.gradle

@@ -27,10 +27,13 @@ dependencies {
     implementation project(':bookkeeper-common-allocator')
     implementation project(':bookkeeper-http:http-server')
     implementation project(':bookkeeper-proto')
+    implementation project(':bookkeeper-slogger:api')
+    implementation project(':bookkeeper-slogger:slf4j')
     implementation project(':bookkeeper-stats')
     implementation project(':bookkeeper-tools-framework')
     implementation project(':circe-checksum')
     implementation project(':cpu-affinity')
+    implementation project(':native-io')
 
     compileOnly depLibs.lombok
     compileOnly depLibs.spotbugsAnnotations

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

@@ -33,7 +33,10 @@ public abstract class AbstractLogCompactor {
     protected final ServerConfiguration conf;
     protected final Throttler throttler;
 
-    interface LogRemovalListener {
+    /**
+     * LogRemovalListener.
+     */
+    public interface LogRemovalListener {
         void removeEntryLog(long logToRemove);
     }
 
@@ -57,7 +60,10 @@ public AbstractLogCompactor(ServerConfiguration conf, LogRemovalListener logRemo
      */
     public void cleanUpAndRecover() {}
 
-    static class Throttler {
+    /**
+     * class Throttler.
+     */
+    public static class Throttler {
         private final RateLimiter rateLimiter;
         private final boolean isThrottleByBytes;
 
@@ -68,7 +74,7 @@ static class Throttler {
         }
 
         // acquire. if bybytes: bytes of this entry; if byentries: 1.
-        void acquire(int permits) {
+        public void acquire(int permits) {
             rateLimiter.acquire(this.isThrottleByBytes ? permits : 1);
         }
     }

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

@@ -28,11 +28,6 @@
  */
 public interface CompactableLedgerStorage extends LedgerStorage {
 
-    /**
-     * @return the EntryLogger used by the ledger storage
-     */
-    EntryLogger getEntryLogger();
-
     /**
      * Get an iterator over a range of ledger ids stored in the bookie.
      *

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

@@ -46,4 +46,14 @@ public long getEntry() {
     public long getLocation() {
         return location;
     }
+
+    @Override
+    public String toString() {
+        return new StringBuilder().append("EntryLocation{")
+            .append("ledger=").append(ledger)
+            .append(",entry=").append(entry)
+            .append(",locationLog=").append(location >> 32 & 0xFFFFFFFF)
+            .append(",locationOffset=").append((int) (location & 0xFFFFFFFF))
+            .append("}").toString();
+    }
 }

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

@@ -27,7 +27,10 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.bookkeeper.bookie.storage.EntryLogScanner;
+import org.apache.bookkeeper.bookie.storage.EntryLoggerIface;
 import org.apache.bookkeeper.conf.ServerConfiguration;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,13 +43,13 @@ public class EntryLogCompactor extends AbstractLogCompactor {
     private static final Logger LOG = LoggerFactory.getLogger(EntryLogCompactor.class);
 
     final CompactionScannerFactory scannerFactory = new CompactionScannerFactory();
-    final EntryLogger entryLogger;
+    final EntryLoggerIface entryLogger;
     final CompactableLedgerStorage ledgerStorage;
     private final int maxOutstandingRequests;
 
     public EntryLogCompactor(
             ServerConfiguration conf,
-            EntryLogger entryLogger,
+            EntryLoggerIface entryLogger,
             CompactableLedgerStorage ledgerStorage,
             LogRemovalListener logRemover) {
         super(conf, logRemover);
@@ -82,9 +85,9 @@ public boolean compact(EntryLogMetadata entryLogMeta) {
     class CompactionScannerFactory {
         List<EntryLocation> offsets = new ArrayList<EntryLocation>();
 
-        EntryLogger.EntryLogScanner newScanner(final EntryLogMetadata meta) {
+        EntryLogScanner newScanner(final EntryLogMetadata meta) {
 
-            return new EntryLogger.EntryLogScanner() {
+            return new EntryLogScanner() {
                 @Override
                 public boolean accept(long ledgerId) {
                     return meta.containsLedger(ledgerId);

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

@@ -22,7 +22,6 @@
 package org.apache.bookkeeper.bookie;
 
 import java.io.Closeable;
-import java.io.IOException;
 import java.util.function.BiConsumer;
 
 import org.apache.bookkeeper.bookie.BookieException.EntryLogMetadataMapException;
@@ -37,7 +36,7 @@ public interface EntryLogMetadataMap extends Closeable {
      *
      * @param entryLogId
      * @return
-     * @throws IOException
+     * @throws EntryLogMetadataMapException
      */
     boolean containsKey(long entryLogId) throws EntryLogMetadataMapException;
 
@@ -46,7 +45,7 @@ public interface EntryLogMetadataMap extends Closeable {
      *
      * @param entryLogId
      * @param entryLogMeta
-     * @throws IOException
+     * @throws EntryLogMetadataMapException
      */
     void put(long entryLogId, EntryLogMetadata entryLogMeta) throws EntryLogMetadataMapException;
 
@@ -55,24 +54,40 @@ public interface EntryLogMetadataMap extends Closeable {
      * have been processed or the action throws an exception.
      *
      * @param action
-     * @throws IOException
+     * @throws EntryLogMetadataMapException
      */
     void forEach(BiConsumer<Long, EntryLogMetadata> action) throws EntryLogMetadataMapException;
 
     /**
      * Removes entryLogMetadata record from the map.
      *
      * @param entryLogId
-     * @throws IOException
+     * @throws EntryLogMetadataMapException
      */
     void remove(long entryLogId) throws EntryLogMetadataMapException;
 
     /**
      * Returns number of entryLogMetadata records presents into the map.
      *
      * @return
-     * @throws IOException
+     * @throws EntryLogMetadataMapException
      */
     int size() throws EntryLogMetadataMapException;
 
+    /**
+     * Returns true iff there are no elements in the map.
+     *
+     * @return
+     */
+    default boolean isEmpty() throws EntryLogMetadataMapException {
+        return size() == 0;
+    }
+
+    /**
+     * Clear all records from the map.
+     * For unit tests.
+     *
+     * @throws EntryLogMetadataMapException
+     */
+    void clear() throws EntryLogMetadataMapException;
 }