[improve][misc] Sync commits from apache into 3.1_ds

buildtools/pom.xml

@@ -47,7 +47,7 @@
     <license-maven-plugin.version>4.1</license-maven-plugin.version>
     <puppycrawl.checkstyle.version>10.14.2</puppycrawl.checkstyle.version>
     <maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
-    <netty.version>4.1.111.Final</netty.version>
+    <netty.version>4.1.113.Final</netty.version>
     <guice.version>4.2.3</guice.version>
     <guava.version>32.1.1-jre</guava.version>
     <ant.version>1.10.12</ant.version>

conf/broker.conf

@@ -463,6 +463,16 @@ dispatcherReadFailureBackoffMaxTimeInMs=60000
 # The read failure backoff mandatory stop time in milliseconds. By default it is 0s.
 dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
 
+# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
+# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
+# delay. This parameter sets the initial backoff delay in milliseconds.
+dispatcherRetryBackoffInitialTimeInMs=100
+
+# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
+# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
+# delay. This parameter sets the maximum backoff delay in milliseconds.
+dispatcherRetryBackoffMaxTimeInMs=1000
+
 # Precise dispathcer flow control according to history message number of each entry
 preciseDispatcherFlowControl=false
 

conf/standalone.conf

@@ -279,6 +279,16 @@ dispatcherReadFailureBackoffMaxTimeInMs=60000
 # The read failure backoff mandatory stop time in milliseconds. By default it is 0s.
 dispatcherReadFailureBackoffMandatoryStopTimeInMs=0
 
+# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
+# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
+# delay. This parameter sets the initial backoff delay in milliseconds.
+dispatcherRetryBackoffInitialTimeInMs=100
+
+# On Shared and KeyShared subscriptions, if all available messages in the subscription are filtered
+# out and not dispatched to any consumer, message dispatching will be rescheduled with a backoff
+# delay. This parameter sets the maximum backoff delay in milliseconds.
+dispatcherRetryBackoffMaxTimeInMs=1000
+
 # Precise dispatcher flow control according to history message number of each entry
 preciseDispatcherFlowControl=false
 

distribution/server/src/assemble/LICENSE.bin.txt

@@ -289,33 +289,33 @@ The Apache Software License, Version 2.0
     - org.apache.commons-commons-lang3-3.11.jar
     - org.apache.commons-commons-text-1.10.0.jar
  * Netty
-    - io.netty-netty-buffer-4.1.111.Final.jar
-    - io.netty-netty-codec-4.1.111.Final.jar
-    - io.netty-netty-codec-dns-4.1.111.Final.jar
-    - io.netty-netty-codec-http-4.1.111.Final.jar
-    - io.netty-netty-codec-http2-4.1.111.Final.jar
-    - io.netty-netty-codec-socks-4.1.111.Final.jar
-    - io.netty-netty-codec-haproxy-4.1.111.Final.jar
-    - io.netty-netty-common-4.1.111.Final.jar
-    - io.netty-netty-handler-4.1.111.Final.jar
-    - io.netty-netty-handler-proxy-4.1.111.Final.jar
-    - io.netty-netty-resolver-4.1.111.Final.jar
-    - io.netty-netty-resolver-dns-4.1.111.Final.jar
-    - io.netty-netty-resolver-dns-classes-macos-4.1.111.Final.jar
-    - io.netty-netty-resolver-dns-native-macos-4.1.111.Final-osx-aarch_64.jar
-    - io.netty-netty-resolver-dns-native-macos-4.1.111.Final-osx-x86_64.jar
-    - io.netty-netty-transport-4.1.111.Final.jar
-    - io.netty-netty-transport-classes-epoll-4.1.111.Final.jar
-    - io.netty-netty-transport-native-epoll-4.1.111.Final-linux-aarch_64.jar
-    - io.netty-netty-transport-native-epoll-4.1.111.Final-linux-x86_64.jar
-    - io.netty-netty-transport-native-unix-common-4.1.111.Final.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-linux-aarch_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-linux-x86_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-osx-aarch_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-osx-x86_64.jar
-    - io.netty-netty-tcnative-boringssl-static-2.0.65.Final-windows-x86_64.jar
-    - io.netty-netty-tcnative-classes-2.0.65.Final.jar
+    - io.netty-netty-buffer-4.1.113.Final.jar
+    - io.netty-netty-codec-4.1.113.Final.jar
+    - io.netty-netty-codec-dns-4.1.113.Final.jar
+    - io.netty-netty-codec-http-4.1.113.Final.jar
+    - io.netty-netty-codec-http2-4.1.113.Final.jar
+    - io.netty-netty-codec-socks-4.1.113.Final.jar
+    - io.netty-netty-codec-haproxy-4.1.113.Final.jar
+    - io.netty-netty-common-4.1.113.Final.jar
+    - io.netty-netty-handler-4.1.113.Final.jar
+    - io.netty-netty-handler-proxy-4.1.113.Final.jar
+    - io.netty-netty-resolver-4.1.113.Final.jar
+    - io.netty-netty-resolver-dns-4.1.113.Final.jar
+    - io.netty-netty-resolver-dns-classes-macos-4.1.113.Final.jar
+    - io.netty-netty-resolver-dns-native-macos-4.1.113.Final-osx-aarch_64.jar
+    - io.netty-netty-resolver-dns-native-macos-4.1.113.Final-osx-x86_64.jar
+    - io.netty-netty-transport-4.1.113.Final.jar
+    - io.netty-netty-transport-classes-epoll-4.1.113.Final.jar
+    - io.netty-netty-transport-native-epoll-4.1.113.Final-linux-aarch_64.jar
+    - io.netty-netty-transport-native-epoll-4.1.113.Final-linux-x86_64.jar
+    - io.netty-netty-transport-native-unix-common-4.1.113.Final.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.66.Final.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.66.Final-linux-aarch_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.66.Final-linux-x86_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.66.Final-osx-aarch_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.66.Final-osx-x86_64.jar
+    - io.netty-netty-tcnative-boringssl-static-2.0.66.Final-windows-x86_64.jar
+    - io.netty-netty-tcnative-classes-2.0.66.Final.jar
     - io.netty.incubator-netty-incubator-transport-classes-io_uring-0.0.21.Final.jar
     - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.21.Final-linux-x86_64.jar
     - io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.21.Final-linux-aarch_64.jar

distribution/shell/src/assemble/LICENSE.bin.txt

@@ -344,35 +344,35 @@ The Apache Software License, Version 2.0
     - commons-text-1.10.0.jar
     - commons-compress-1.26.0.jar
  * Netty
-    - netty-buffer-4.1.111.Final.jar
-    - netty-codec-4.1.111.Final.jar
-    - netty-codec-dns-4.1.111.Final.jar
-    - netty-codec-http-4.1.111.Final.jar
-    - netty-codec-socks-4.1.111.Final.jar
-    - netty-codec-haproxy-4.1.111.Final.jar
-    - netty-common-4.1.111.Final.jar
-    - netty-handler-4.1.111.Final.jar
-    - netty-handler-proxy-4.1.111.Final.jar
-    - netty-resolver-4.1.111.Final.jar
-    - netty-resolver-dns-4.1.111.Final.jar
-    - netty-transport-4.1.111.Final.jar
-    - netty-transport-classes-epoll-4.1.111.Final.jar
-    - netty-transport-native-epoll-4.1.111.Final-linux-aarch_64.jar
-    - netty-transport-native-epoll-4.1.111.Final-linux-x86_64.jar
-    - netty-transport-native-unix-common-4.1.111.Final.jar
-    - netty-tcnative-boringssl-static-2.0.65.Final.jar
-    - netty-tcnative-boringssl-static-2.0.65.Final-linux-aarch_64.jar
-    - netty-tcnative-boringssl-static-2.0.65.Final-linux-x86_64.jar
-    - netty-tcnative-boringssl-static-2.0.65.Final-osx-aarch_64.jar
-    - netty-tcnative-boringssl-static-2.0.65.Final-osx-x86_64.jar
-    - netty-tcnative-boringssl-static-2.0.65.Final-windows-x86_64.jar
-    - netty-tcnative-classes-2.0.65.Final.jar
+    - netty-buffer-4.1.113.Final.jar
+    - netty-codec-4.1.113.Final.jar
+    - netty-codec-dns-4.1.113.Final.jar
+    - netty-codec-http-4.1.113.Final.jar
+    - netty-codec-socks-4.1.113.Final.jar
+    - netty-codec-haproxy-4.1.113.Final.jar
+    - netty-common-4.1.113.Final.jar
+    - netty-handler-4.1.113.Final.jar
+    - netty-handler-proxy-4.1.113.Final.jar
+    - netty-resolver-4.1.113.Final.jar
+    - netty-resolver-dns-4.1.113.Final.jar
+    - netty-transport-4.1.113.Final.jar
+    - netty-transport-classes-epoll-4.1.113.Final.jar
+    - netty-transport-native-epoll-4.1.113.Final-linux-aarch_64.jar
+    - netty-transport-native-epoll-4.1.113.Final-linux-x86_64.jar
+    - netty-transport-native-unix-common-4.1.113.Final.jar
+    - netty-tcnative-boringssl-static-2.0.66.Final.jar
+    - netty-tcnative-boringssl-static-2.0.66.Final-linux-aarch_64.jar
+    - netty-tcnative-boringssl-static-2.0.66.Final-linux-x86_64.jar
+    - netty-tcnative-boringssl-static-2.0.66.Final-osx-aarch_64.jar
+    - netty-tcnative-boringssl-static-2.0.66.Final-osx-x86_64.jar
+    - netty-tcnative-boringssl-static-2.0.66.Final-windows-x86_64.jar
+    - netty-tcnative-classes-2.0.66.Final.jar
     - netty-incubator-transport-classes-io_uring-0.0.21.Final.jar
     - netty-incubator-transport-native-io_uring-0.0.21.Final-linux-aarch_64.jar
     - netty-incubator-transport-native-io_uring-0.0.21.Final-linux-x86_64.jar
-    - netty-resolver-dns-classes-macos-4.1.111.Final.jar
-    - netty-resolver-dns-native-macos-4.1.111.Final-osx-aarch_64.jar
-    - netty-resolver-dns-native-macos-4.1.111.Final-osx-x86_64.jar
+    - netty-resolver-dns-classes-macos-4.1.113.Final.jar
+    - netty-resolver-dns-native-macos-4.1.113.Final-osx-aarch_64.jar
+    - netty-resolver-dns-native-macos-4.1.113.Final-osx-x86_64.jar
  * Prometheus client
     - simpleclient-0.16.0.jar
     - simpleclient_log4j2-0.16.0.jar

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java

@@ -230,5 +230,9 @@ default void scanLedgers(OffloadedLedgerMetadataConsumer consumer,
                              Map<String, String> offloadDriverMetadata) throws ManagedLedgerException {
         throw ManagedLedgerException.getManagedLedgerException(new UnsupportedOperationException());
     }
+
+    default boolean isAppendable() {
+        return true;
+    }
 }
 

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -94,6 +94,7 @@
 import org.apache.bookkeeper.mledger.AsyncCallbacks.TerminateCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.UpdatePropertiesCallback;
 import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedLedger;
 import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
@@ -2494,9 +2495,8 @@ private void scheduleDeferredTrimming(boolean isTruncate, CompletableFuture<?> p
                 100, TimeUnit.MILLISECONDS);
     }
 
-    private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
-        if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
-                || config.getLedgerOffloader().getOffloadPolicies() == null) {
+    public void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
+        if (getOffloadPoliciesIfAppendable().isEmpty()) {
             return;
         }
 
@@ -2512,8 +2512,7 @@ private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
 
     private void maybeOffload(long offloadThresholdInBytes, long offloadThresholdInSeconds,
                               CompletableFuture<PositionImpl> finalPromise) {
-        if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
-                || config.getLedgerOffloader().getOffloadPolicies() == null) {
+        if (getOffloadPoliciesIfAppendable().isEmpty()) {
             String msg = String.format("[%s] Nothing to offload due to offloader or offloadPolicies is NULL", name);
             finalPromise.completeExceptionally(new IllegalArgumentException(msg));
             return;
@@ -2615,6 +2614,16 @@ void internalTrimConsumedLedgers(CompletableFuture<?> promise) {
         internalTrimLedgers(false, promise);
     }
 
+    private Optional<OffloadPolicies> getOffloadPoliciesIfAppendable() {
+        LedgerOffloader ledgerOffloader = config.getLedgerOffloader();
+        if (ledgerOffloader == null
+                || !ledgerOffloader.isAppendable()
+                || ledgerOffloader.getOffloadPolicies() == null) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(ledgerOffloader.getOffloadPolicies());
+    }
+
     @VisibleForTesting
     List<Long> internalEvictOffloadedLedgers() {
         int inactiveOffloadedLedgerEvictionTimeMs = config.getInactiveOffloadedLedgerEvictionTimeMs();
@@ -2668,10 +2677,7 @@ void internalTrimLedgers(boolean isTruncate, CompletableFuture<?> promise) {
 
         List<LedgerInfo> ledgersToDelete = new ArrayList<>();
         List<LedgerInfo> offloadedLedgersToDelete = new ArrayList<>();
-        Optional<OffloadPolicies> optionalOffloadPolicies = Optional.ofNullable(config.getLedgerOffloader() != null
-                && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
-                ? config.getLedgerOffloader().getOffloadPolicies()
-                : null);
+        Optional<OffloadPolicies> optionalOffloadPolicies = getOffloadPoliciesIfAppendable();
         synchronized (this) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] Start TrimConsumedLedgers. ledgers={} totalSize={}", name, ledgers.keySet(),
@@ -3198,8 +3204,10 @@ public void offloadFailed(ManagedLedgerException e, Object ctx) {
 
     @Override
     public void asyncOffloadPrefix(Position pos, OffloadCallback callback, Object ctx) {
-        if (config.getLedgerOffloader() != null && config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE) {
-            callback.offloadFailed(new ManagedLedgerException("NullLedgerOffloader"), ctx);
+        LedgerOffloader ledgerOffloader = config.getLedgerOffloader();
+        if (ledgerOffloader != null && !ledgerOffloader.isAppendable()) {
+            String msg = String.format("[%s] does not support offload", ledgerOffloader.getClass().getSimpleName());
+            callback.offloadFailed(new ManagedLedgerException(msg), ctx);
             return;
         }
         PositionImpl requestOffloadTo = (PositionImpl) pos;

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonAppendableLedgerOffloader.java

@@ -0,0 +1,74 @@
+/*
+ * 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.mledger.impl;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
+import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
+import org.apache.pulsar.common.util.FutureUtil;
+
+public class NonAppendableLedgerOffloader implements LedgerOffloader {
+    private LedgerOffloader delegate;
+
+    public NonAppendableLedgerOffloader(LedgerOffloader delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public String getOffloadDriverName() {
+        return delegate.getOffloadDriverName();
+    }
+
+    @Override
+    public CompletableFuture<Void> offload(ReadHandle ledger,
+                                           UUID uid,
+                                           Map<String, String> extraMetadata) {
+        return FutureUtil.failedFuture(new UnsupportedOperationException());
+    }
+
+    @Override
+    public CompletableFuture<ReadHandle> readOffloaded(long ledgerId, UUID uid,
+                                                       Map<String, String> offloadDriverMetadata) {
+        return delegate.readOffloaded(ledgerId, uid, offloadDriverMetadata);
+    }
+
+    @Override
+    public CompletableFuture<Void> deleteOffloaded(long ledgerId, UUID uid,
+                                                   Map<String, String> offloadDriverMetadata) {
+        return delegate.deleteOffloaded(ledgerId, uid, offloadDriverMetadata);
+    }
+
+    @Override
+    public OffloadPoliciesImpl getOffloadPolicies() {
+        return delegate.getOffloadPolicies();
+    }
+
+    @Override
+    public void close() {
+        delegate.close();
+    }
+
+    @Override
+    public boolean isAppendable() {
+        return false;
+    }
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NullLedgerOffloader.java

@@ -70,4 +70,9 @@ public OffloadPoliciesImpl getOffloadPolicies() {
     public void close() {
 
     }
+
+    @Override
+    public boolean isAppendable() {
+        return false;
+    }
 }

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -3849,7 +3849,7 @@ public void testDoNotGetOffloadPoliciesMultipleTimesWhenTrimLedgers() throws Exc
         config.setLedgerOffloader(ledgerOffloader);
 
         ledger.internalTrimConsumedLedgers(Futures.NULL_PROMISE);
-        verify(ledgerOffloader, times(1)).getOffloadPolicies();
+        verify(ledgerOffloader, times(1)).isAppendable();
     }
 
     @Test(timeOut = 30000)