This is an automated email from the ASF dual-hosted git repository.
chenhang pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new 97a7b70d731 [improve] [broker] Trigger offload on topic load (#22652)
97a7b70d731 is described below
commit 97a7b70d7310fded1d5a16be36df92aa882c935c
Author: Hang Chen <[email protected]>
AuthorDate: Fri May 17 21:59:42 2024 +0800
[improve] [broker] Trigger offload on topic load (#22652)
(cherry picked from commit 400a286ca09d4f45f388616fd996f0605f19e5a4)
---
conf/broker.conf | 3 +++
conf/standalone.conf | 3 +++
.../apache/bookkeeper/mledger/ManagedLedgerConfig.java | 17 +++++++++++++++++
.../mledger/impl/ManagedLedgerFactoryImpl.java | 5 +++++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 4 ++--
.../org/apache/pulsar/broker/ServiceConfiguration.java | 9 +++++++--
.../org/apache/pulsar/broker/service/BrokerService.java | 1 +
7 files changed, 38 insertions(+), 4 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 1ef68a0395c..2a9641b5b90 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1211,6 +1211,9 @@ managedLedgerDataReadPriority=tiered-storage-first
# (default is -1, which is disabled)
managedLedgerOffloadThresholdInSeconds=-1
+# Trigger offload on topic load or not. Default is false.
+# triggerOffloadOnTopicLoad=false
+
# Max number of entries to append to a cursor ledger
managedLedgerCursorMaxEntriesPerLedger=50000
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 758a79d0fa1..1a0c501899d 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -835,6 +835,9 @@ managedLedgerPrometheusStatsLatencyRolloverSeconds=60
# Whether trace managed ledger task execution time
managedLedgerTraceTaskExecution=true
+# Trigger offload on topic load or not. Default is false.
+# triggerOffloadOnTopicLoad=false
+
# If you want to custom bookie ID or use a dynamic network address for the
bookie,
# you can set this option.
# Bookie advertises itself using bookieId rather than
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index a097d41539f..03439f93cca 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -85,6 +85,7 @@ public class ManagedLedgerConfig {
private int minimumBacklogCursorsForCaching = 0;
private int minimumBacklogEntriesForCaching = 1000;
private int maxBacklogBetweenCursorsForCaching = 1000;
+ private boolean triggerOffloadOnTopicLoad = false;
@Getter
@Setter
@@ -748,6 +749,22 @@ public class ManagedLedgerConfig {
this.maxBacklogBetweenCursorsForCaching =
maxBacklogBetweenCursorsForCaching;
}
+ /**
+ * Trigger offload on topic load.
+ * @return
+ */
+ public boolean isTriggerOffloadOnTopicLoad() {
+ return triggerOffloadOnTopicLoad;
+ }
+
+ /**
+ * Set trigger offload on topic load.
+ * @param triggerOffloadOnTopicLoad
+ */
+ public void setTriggerOffloadOnTopicLoad(boolean
triggerOffloadOnTopicLoad) {
+ this.triggerOffloadOnTopicLoad = triggerOffloadOnTopicLoad;
+ }
+
public String getShadowSource() {
return MapUtils.getString(properties, PROPERTY_SOURCE_TOPIC_KEY);
}
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
index a0929044a6a..ed803a81462 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.mledger.impl;
import static com.google.common.base.Preconditions.checkArgument;
import static
org.apache.bookkeeper.mledger.ManagedLedgerException.getManagedLedgerException;
+import static
org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.NULL_OFFLOAD_PROMISE;
import static
org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
import com.google.common.base.Predicates;
import com.google.common.collect.Maps;
@@ -397,6 +398,10 @@ public class ManagedLedgerFactoryImpl implements
ManagedLedgerFactory {
// May need to update the cursor position
newledger.maybeUpdateCursorBeforeTrimmingConsumedLedger();
+ // May need to trigger offloading
+ if (config.isTriggerOffloadOnTopicLoad()) {
+
newledger.maybeOffloadInBackground(NULL_OFFLOAD_PROMISE);
+ }
}
@Override
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 1e3e7e065d7..b23759021d2 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -213,7 +213,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
private final CallbackMutex trimmerMutex = new CallbackMutex();
private final CallbackMutex offloadMutex = new CallbackMutex();
- private static final CompletableFuture<PositionImpl> NULL_OFFLOAD_PROMISE
= CompletableFuture
+ public static final CompletableFuture<PositionImpl> NULL_OFFLOAD_PROMISE =
CompletableFuture
.completedFuture(PositionImpl.LATEST);
protected volatile LedgerHandle currentLedger;
protected volatile long currentLedgerEntries = 0;
@@ -2483,7 +2483,7 @@ public class ManagedLedgerImpl implements ManagedLedger,
CreateCallback {
100, TimeUnit.MILLISECONDS);
}
- private void maybeOffloadInBackground(CompletableFuture<PositionImpl>
promise) {
+ public void maybeOffloadInBackground(CompletableFuture<PositionImpl>
promise) {
if (config.getLedgerOffloader() == null || config.getLedgerOffloader()
== NullLedgerOffloader.INSTANCE
|| config.getLedgerOffloader().getOffloadPolicies() == null) {
return;
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 11d1d663f42..6d4635681ed 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2096,10 +2096,15 @@ public class ServiceConfiguration implements
PulsarConfiguration {
)
private long managedLedgerOffloadAutoTriggerSizeThresholdBytes = -1L;
@FieldContext(
- category = CATEGORY_STORAGE_OFFLOADING,
- doc = "The threshold to triggering automatic offload to long term
storage"
+ category = CATEGORY_STORAGE_OFFLOADING,
+ doc = "The threshold to triggering automatic offload to long term
storage"
)
private long managedLedgerOffloadThresholdInSeconds = -1L;
+ @FieldContext(
+ category = CATEGORY_STORAGE_OFFLOADING,
+ doc = "Trigger offload on topic load or not. Default is false"
+ )
+ private boolean triggerOffloadOnTopicLoad = false;
@FieldContext(
category = CATEGORY_STORAGE_ML,
doc = "Max number of entries to append to a cursor ledger"
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b2c36adeace..b46879f3866 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -1985,6 +1985,7 @@ public class BrokerService implements Closeable {
.setLedgerOffloader(pulsar.getManagedLedgerOffloader(namespace,
offloadPolicies));
}
}
+
managedLedgerConfig.setTriggerOffloadOnTopicLoad(serviceConfig.isTriggerOffloadOnTopicLoad());
managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled(
serviceConfig.isAcknowledgmentAtBatchIndexLevelEnabled());
