You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xi...@apache.org on 2022/12/13 09:18:13 UTC
[pulsar] 02/04: [improve][broker]Improve PersistentMessageExpiryMonitor expire speed when ledger not existed (#17842)
This is an automated email from the ASF dual-hosted git repository.
xiangying pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fcc98d1810449cdab964c942a65de6a33f5bd1ca
Author: Xiaoyu Hou <ho...@apache.org>
AuthorDate: Mon Oct 3 23:13:03 2022 +0800
[improve][broker]Improve PersistentMessageExpiryMonitor expire speed when ledger not existed (#17842)
(cherry picked from commit af11c32611e41d708e76264d47ef9501534c6c5c)
---
.../apache/bookkeeper/mledger/ManagedLedger.java | 7 ++++++
.../bookkeeper/mledger/ManagedLedgerException.java | 6 ++++++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 25 ++++++++++++++++++++--
.../persistent/PersistentMessageExpiryMonitor.java | 25 +++++++++++++++++++++-
.../offload/jcloud/impl/MockManagedLedger.java | 7 ++++++
5 files changed, 67 insertions(+), 3 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
index 7196a3b4c03..ba33b5e43e8 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.mledger;
import io.netty.buffer.ByteBuf;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.common.annotation.InterfaceAudience;
import org.apache.bookkeeper.common.annotation.InterfaceStability;
@@ -648,6 +649,12 @@ public interface ManagedLedger {
*/
CompletableFuture<LedgerInfo> getLedgerInfo(long ledgerId);
+ /**
+ * Get basic ledger summary.
+ * will get {@link Optional#empty()} if corresponding ledger not exists.
+ */
+ Optional<LedgerInfo> getOptionalLedgerInfo(long ledgerId);
+
/**
* Truncate ledgers
* The truncate operation will move all cursors to the end of the topic and delete all inactive ledgers.
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java
index 3dd4b9e3ca1..5e712cf9e81 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerException.java
@@ -147,6 +147,12 @@ public class ManagedLedgerException extends Exception {
}
}
+ public static class LedgerNotExistException extends NonRecoverableLedgerException {
+ public LedgerNotExistException(String msg) {
+ super(msg);
+ }
+ }
+
public static class InvalidReplayPositionException extends ManagedLedgerException {
public InvalidReplayPositionException(String msg) {
super(msg);
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 ebe64298b71..8725a3127b1 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
@@ -101,6 +101,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.BadVersionException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.CursorNotFoundException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.InvalidCursorPositionException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.LedgerNotExistException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerAlreadyClosedException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerFencedException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.ManagedLedgerInterceptException;
@@ -1813,6 +1814,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
return result;
}
+ @Override
+ public Optional<LedgerInfo> getOptionalLedgerInfo(long ledgerId) {
+ return Optional.ofNullable(ledgers.get(ledgerId));
+ }
+
CompletableFuture<ReadHandle> getLedgerHandle(long ledgerId) {
CompletableFuture<ReadHandle> ledgerHandle = ledgerCache.get(ledgerId);
if (ledgerHandle != null) {
@@ -1919,7 +1925,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
} else {
log.error("[{}] Failed to get message with ledger {}:{} the ledgerId does not belong to this topic "
+ "or has been deleted.", name, position.getLedgerId(), position.getEntryId());
- callback.readEntryFailed(new ManagedLedgerException.NonRecoverableLedgerException("Message not found, "
+ callback.readEntryFailed(new LedgerNotExistException("Message not found, "
+ "the ledgerId does not belong to this topic or has been deleted"), ctx);
}
@@ -3744,11 +3750,26 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}
+ private static boolean isLedgerNotExistException(int rc) {
+ switch (rc) {
+ case Code.NoSuchLedgerExistsException:
+ case Code.NoSuchLedgerExistsOnMetadataServerException:
+ return true;
+
+ default:
+ return false;
+ }
+ }
+
public static ManagedLedgerException createManagedLedgerException(int bkErrorCode) {
if (bkErrorCode == BKException.Code.TooManyRequestsException) {
return new TooManyRequestsException("Too many request error from bookies");
} else if (isBkErrorNotRecoverable(bkErrorCode)) {
- return new NonRecoverableLedgerException(BKException.getMessage(bkErrorCode));
+ if (isLedgerNotExistException(bkErrorCode)) {
+ return new LedgerNotExistException(BKException.getMessage(bkErrorCode));
+ } else {
+ return new NonRecoverableLedgerException(BKException.getMessage(bkErrorCode));
+ }
} else {
return new ManagedLedgerException(BKException.getMessage(bkErrorCode));
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
index 8aabc7bf3d8..9b1cd75282a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -26,8 +26,10 @@ import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerException.LedgerNotExistException;
import org.apache.bookkeeper.mledger.ManagedLedgerException.NonRecoverableLedgerException;
import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
@@ -191,7 +193,28 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback {
&& (exception instanceof NonRecoverableLedgerException)) {
log.warn("[{}][{}] read failed from ledger at position:{} : {}", topicName, subName, failedReadPosition,
exception.getMessage());
- findEntryComplete(failedReadPosition.get(), ctx);
+ if (exception instanceof LedgerNotExistException) {
+ long failedLedgerId = failedReadPosition.get().getLedgerId();
+ ManagedLedgerImpl ledger = ((ManagedLedgerImpl) cursor.getManagedLedger());
+ Position lastPositionInLedger = ledger.getOptionalLedgerInfo(failedLedgerId)
+ .map(ledgerInfo -> PositionImpl.get(failedLedgerId, ledgerInfo.getEntries() - 1))
+ .orElseGet(() -> {
+ Long nextExistingLedger = ledger.getNextValidLedger(failedReadPosition.get().getLedgerId());
+ if (nextExistingLedger == null) {
+ log.info("[{}] [{}] Couldn't find next next valid ledger for expiry monitor when find "
+ + "entry failed {}", ledger.getName(), ledger.getName(),
+ failedReadPosition);
+ return (PositionImpl) failedReadPosition.get();
+ } else {
+ return PositionImpl.get(nextExistingLedger, -1);
+ }
+ });
+ log.info("[{}][{}] ledger not existed, will complete the last position of the non-existed"
+ + " ledger:{}", topicName, subName, lastPositionInLedger);
+ findEntryComplete(lastPositionInLedger, ctx);
+ } else {
+ findEntryComplete(failedReadPosition.get(), ctx);
+ }
}
expirationCheckInProgress = FALSE;
updateRates();
diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
index 3eaf276c3c5..1f7a7b376c5 100644
--- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
+++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/MockManagedLedger.java
@@ -21,6 +21,7 @@ package org.apache.bookkeeper.mledger.offload.jcloud.impl;
import com.google.common.base.Predicate;
import io.netty.buffer.ByteBuf;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
@@ -354,6 +355,12 @@ public class MockManagedLedger implements ManagedLedger {
return CompletableFuture.completedFuture(build);
}
+ @Override
+ public Optional<LedgerInfo> getOptionalLedgerInfo(long ledgerId) {
+ final LedgerInfo build = LedgerInfo.newBuilder().setLedgerId(ledgerId).setSize(100).setEntries(20).build();
+ return Optional.of(build);
+ }
+
@Override
public CompletableFuture<Void> asyncTruncate() {
return CompletableFuture.completedFuture(null);