[improve][misc] Sync commits from apache/branch-4.16 into ds-4.16

.github/actions/tune-runner-vm/action.yml

@@ -76,13 +76,6 @@ runs:
 
             # stop Azure Linux agent to save RAM
             sudo systemctl stop walinuxagent.service || true
-
-            # enable docker experimental mode which is
-            # required for using "docker build --squash" / "-Ddocker.squash=true"
-            daemon_json="$(sudo cat /etc/docker/daemon.json  | jq '.experimental = true')"
-            echo "$daemon_json" | sudo tee /etc/docker/daemon.json
-            # restart docker daemon
-            sudo systemctl restart docker
             echo '::endgroup::'
 
             # show memory

.github/workflows/bk-ci.yml

@@ -24,14 +24,17 @@ on:
     branches:
       - master
       - branch-*
+      - ds-*
   push:
     branches:
       - master
       - branch-*
+      - ds-*
   workflow_dispatch:
 
 env:
   MAVEN_OPTS: -Xss1500k -Xmx1500m -Daether.connector.http.reuseConnections=false -Daether.connector.requestTimeout=60000 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000
+  NIST_NVD_API_KEY: ${{ secrets.NIST_NVD_API_KEY }}
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
@@ -154,6 +157,13 @@ jobs:
           distribution: 'temurin'
           java-version: 11
 
+      - name: Tune Java DNS TTL settings
+        run: |
+          sudo tee -a $JAVA_HOME/conf/security/java.security <<EOF
+          networkaddress.cache.ttl=1
+          networkaddress.cache.negative.ttl=1
+          EOF
+
       - name: Build
         run: |
           projects_list=
@@ -228,6 +238,13 @@ jobs:
           distribution: 'temurin'
           java-version: 11
 
+      - name: Tune Java DNS TTL settings
+        run: |
+          sudo tee -a $JAVA_HOME/conf/security/java.security <<EOF
+          networkaddress.cache.ttl=1
+          networkaddress.cache.negative.ttl=1
+          EOF
+
       - name: Pick ubuntu mirror for the docker image build
         run: |
           # pick the closest ubuntu mirror and set it to UBUNTU_MIRROR environment variable

.github/workflows/bk-streamstorage-python.yml

@@ -23,13 +23,15 @@ on:
     branches:
       - master
       - branch-*
+      - ds-*
     paths:
       - 'stream/**'
       - '.github/workflows/bk-streamstorage-python.yml'
   push:
     branches:
       - master
       - branch-*
+      - ds-*
     paths:
       - 'stream/**'
       - '.github/workflows/bk-streamstorage-python.yml'

.github/workflows/dead-link-checker.yaml

@@ -22,6 +22,7 @@ on:
     branches:
       - master
       - branch-*
+      - ds-*
     paths:
       - '**.md'
 

.github/workflows/website-pr-validation.yml

@@ -25,6 +25,7 @@ on:
     branches:
       - master
       - branch-*
+      - ds-*
     paths:
       - 'site3/**'
       - '.github/workflows/website-pr-validation.yaml'

bin/bkctl

@@ -28,7 +28,7 @@ source ${BK_HOME}/bin/common.sh
 source ${BK_HOME}/conf/bk_cli_env.sh
 
 CLI_MODULE_PATH=tools/all
-CLI_MODULE_NAME="(org.apache.bookkeeper-)?bookkeeper-tools"
+CLI_MODULE_NAME="(com.datastax.oss-)?bookkeeper-tools"
 CLI_MODULE_HOME=${BK_HOME}/${CLI_MODULE_PATH}
 
 # find the module jar

bin/bkperf

@@ -27,7 +27,7 @@ source ${BK_HOME}/bin/common.sh
 source ${BK_HOME}/conf/bk_cli_env.sh
 
 CLI_MODULE_PATH=tools/perf
-CLI_MODULE_NAME="(org.apache.bookkeeper-)?bookkeeper-perf"
+CLI_MODULE_NAME="(com.datastax.oss-)?bookkeeper-perf"
 CLI_MODULE_HOME=${BK_HOME}/${CLI_MODULE_PATH}
 
 # find the module jar

bin/common.sh

@@ -133,8 +133,8 @@ CLI_GC_OPTS=${CLI_GC_OPTS:-"${DEFAULT_CLI_GC_OPTS}"}
 CLI_GC_LOGGING_OPTS=${CLI_GC_LOGGING_OPTS:-"${DEFAULT_CLI_GC_LOGGING_OPTS}"}
 
 # module names
-BOOKIE_SERVER_MODULE_NAME="(org.apache.bookkeeper-)?bookkeeper-server"
-TABLE_SERVICE_MODULE_NAME="(org.apache.bookkeeper-)?stream-storage-server"
+BOOKIE_SERVER_MODULE_NAME="(com.datastax.oss-)?bookkeeper-server"
+TABLE_SERVICE_MODULE_NAME="(com.datastax.oss-)?stream-storage-server"
 
 is_released_binary() {
   if [ -d ${BK_HOME}/lib ]; then

bin/dlog

@@ -25,7 +25,7 @@ BK_HOME=`cd ${BINDIR}/..;pwd`
 
 source ${BK_HOME}/bin/common.sh
 
-DLOG_MODULE_NAME="(org.apache.distributedlog-)?distributedlog-core"
+DLOG_MODULE_NAME="(com.datastax.oss-)?distributedlog-core"
 DLOG_MODULE_PATH=stream/distributedlog/core
 DLOG_MODULE_HOME=${BK_HOME}/${DLOG_MODULE_PATH}
 

bookkeeper-common-allocator/pom.xml

@@ -29,6 +29,16 @@
       <groupId>io.netty</groupId>
       <artifactId>netty-buffer</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-inline</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/ByteBufAllocatorBuilder.java

@@ -92,4 +92,6 @@ static ByteBufAllocatorBuilder create() {
      * <p>Default is {@link LeakDetectionPolicy#Disabled}
      */
     ByteBufAllocatorBuilder leakDetectionPolicy(LeakDetectionPolicy leakDetectionPolicy);
+
+    ByteBufAllocatorBuilder exitOnOutOfMemory(boolean exitOnOutOfMemory);
 }

bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/impl/ByteBufAllocatorBuilderImpl.java

@@ -37,11 +37,12 @@ public class ByteBufAllocatorBuilderImpl implements ByteBufAllocatorBuilder {
     OutOfMemoryPolicy outOfMemoryPolicy = OutOfMemoryPolicy.FallbackToHeap;
     Consumer<OutOfMemoryError> outOfMemoryListener = null;
     LeakDetectionPolicy leakDetectionPolicy = LeakDetectionPolicy.Disabled;
+    boolean exitOnOutOfMemory = false;
 
     @Override
     public ByteBufAllocatorWithOomHandler build() {
         return new ByteBufAllocatorImpl(pooledAllocator, unpooledAllocator, poolingPolicy, poolingConcurrency,
-                outOfMemoryPolicy, outOfMemoryListener, leakDetectionPolicy);
+                outOfMemoryPolicy, outOfMemoryListener, leakDetectionPolicy, exitOnOutOfMemory);
     }
 
     @Override
@@ -86,4 +87,10 @@ public ByteBufAllocatorBuilder leakDetectionPolicy(LeakDetectionPolicy leakDetec
         return this;
     }
 
+    @Override
+    public ByteBufAllocatorBuilder exitOnOutOfMemory(boolean exitOnOutOfMemory) {
+        this.exitOnOutOfMemory = exitOnOutOfMemory;
+        return this;
+    }
+
 }

bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/impl/ByteBufAllocatorImpl.java

@@ -29,6 +29,7 @@
 import org.apache.bookkeeper.common.allocator.LeakDetectionPolicy;
 import org.apache.bookkeeper.common.allocator.OutOfMemoryPolicy;
 import org.apache.bookkeeper.common.allocator.PoolingPolicy;
+import org.apache.bookkeeper.common.util.ShutdownUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -48,15 +49,24 @@ public class ByteBufAllocatorImpl extends AbstractByteBufAllocator implements By
     private final PoolingPolicy poolingPolicy;
     private final OutOfMemoryPolicy outOfMemoryPolicy;
     private Consumer<OutOfMemoryError> outOfMemoryListener;
+    private final boolean exitOnOutOfMemory;
 
     ByteBufAllocatorImpl(ByteBufAllocator pooledAllocator, ByteBufAllocator unpooledAllocator,
             PoolingPolicy poolingPolicy, int poolingConcurrency, OutOfMemoryPolicy outOfMemoryPolicy,
             Consumer<OutOfMemoryError> outOfMemoryListener,
             LeakDetectionPolicy leakDetectionPolicy) {
-        super(poolingPolicy == PoolingPolicy.PooledDirect /* preferDirect */);
+        this(pooledAllocator, unpooledAllocator, poolingPolicy, poolingConcurrency, outOfMemoryPolicy,
+                outOfMemoryListener, leakDetectionPolicy, false);
+    }
 
+    ByteBufAllocatorImpl(ByteBufAllocator pooledAllocator, ByteBufAllocator unpooledAllocator,
+            PoolingPolicy poolingPolicy, int poolingConcurrency, OutOfMemoryPolicy outOfMemoryPolicy,
+            Consumer<OutOfMemoryError> outOfMemoryListener,
+            LeakDetectionPolicy leakDetectionPolicy, boolean exitOnOutOfMemory) {
+        super(poolingPolicy == PoolingPolicy.PooledDirect /* preferDirect */);
         this.poolingPolicy = poolingPolicy;
         this.outOfMemoryPolicy = outOfMemoryPolicy;
+        this.exitOnOutOfMemory = exitOnOutOfMemory;
         if (outOfMemoryListener == null) {
             this.outOfMemoryListener = (v) -> {
                 log.error("Unable to allocate memory", v);
@@ -146,7 +156,7 @@ protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
                     : unpooledAllocator;
             return alloc.heapBuffer(initialCapacity, maxCapacity);
         } catch (OutOfMemoryError e) {
-            outOfMemoryListener.accept(e);
+            consumeOOMError(e);
             throw e;
         }
     }
@@ -166,12 +176,12 @@ private ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity, boolean ca
                     try {
                         return unpooledAllocator.heapBuffer(initialCapacity, maxCapacity);
                     } catch (OutOfMemoryError e2) {
-                        outOfMemoryListener.accept(e2);
+                        consumeOOMError(e2);
                         throw e2;
                     }
                 } else {
                     // ThrowException
-                    outOfMemoryListener.accept(e);
+                    consumeOOMError(e);
                     throw e;
                 }
             }
@@ -181,12 +191,24 @@ private ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity, boolean ca
             try {
                 return unpooledAllocator.directBuffer(initialCapacity, maxCapacity);
             } catch (OutOfMemoryError e) {
-                outOfMemoryListener.accept(e);
-                throw e;
+                consumeOOMError(e);
+            throw e;
             }
         }
     }
 
+    private void consumeOOMError(OutOfMemoryError outOfMemoryError) {
+        try {
+            outOfMemoryListener.accept(outOfMemoryError);
+        } catch (Throwable e) {
+            log.warn("Consume outOfMemory error failed.", e);
+        }
+        if (exitOnOutOfMemory) {
+            log.info("Exiting JVM process for OOM error: {}", outOfMemoryError.getMessage(), outOfMemoryError);
+            ShutdownUtil.triggerImmediateForcefulShutdown();
+        }
+    }
+
     @Override
     public boolean isDirectBufferPooled() {
         return pooledAllocator != null && pooledAllocator.isDirectBufferPooled();

bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/util/ShutdownUtil.java

@@ -0,0 +1,86 @@
+/*
+ * 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 java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Forked from <a href="https://github.com/apache/pulsar">Pulsar</a>.
+ */
+@Slf4j
+public class ShutdownUtil {
+    private static final Method log4j2ShutdownMethod;
+
+    static {
+        // use reflection to find org.apache.logging.log4j.LogManager.shutdown method
+        Method shutdownMethod = null;
+        try {
+            shutdownMethod = Class.forName("org.apache.logging.log4j.LogManager")
+                    .getMethod("shutdown");
+        } catch (ClassNotFoundException | NoSuchMethodException e) {
+            // ignore when Log4j2 isn't found, log at debug level
+            log.debug("Cannot find org.apache.logging.log4j.LogManager.shutdown method", e);
+        }
+        log4j2ShutdownMethod = shutdownMethod;
+    }
+
+    /**
+     * Triggers an immediate forceful shutdown of the current process.
+     *
+     * @param status Termination status. By convention, a nonzero status code indicates abnormal termination.
+     * @see Runtime#halt(int)
+     */
+    public static void triggerImmediateForcefulShutdown(int status) {
+        triggerImmediateForcefulShutdown(status, true);
+    }
+    public static void triggerImmediateForcefulShutdown(int status, boolean logging) {
+        try {
+            if (status != 0 && logging) {
+                log.warn("Triggering immediate shutdown of current process with status {}", status,
+                        new Exception("Stacktrace for immediate shutdown"));
+            }
+            shutdownLogging();
+        } finally {
+            Runtime.getRuntime().halt(status);
+        }
+    }
+
+    private static void shutdownLogging() {
+        // flush log buffers and shutdown log4j2 logging to prevent log truncation
+        if (log4j2ShutdownMethod != null) {
+            try {
+                // use reflection to call org.apache.logging.log4j.LogManager.shutdown()
+                log4j2ShutdownMethod.invoke(null);
+            } catch (IllegalAccessException | InvocationTargetException e) {
+                log.error("Unable to call org.apache.logging.log4j.LogManager.shutdown using reflection.", e);
+            }
+        }
+    }
+
+    /**
+     * Triggers an immediate forceful shutdown of the current process using 1 as the status code.
+     *
+     * @see Runtime#halt(int)
+     */
+    public static void triggerImmediateForcefulShutdown() {
+        triggerImmediateForcefulShutdown(1);
+    }
+}

bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/util/package-info.java

@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * defines the utilities for allocator used across the project.
+ */
+package org.apache.bookkeeper.common.util;