You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/06/09 14:50:34 UTC
(pulsar) branch branch-4.0 updated (227f947f148 -> 9adbbbf8f39)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 227f947f148 [improve][offloaders] Automatically evict Offloaded Ledgers from memory (#19783)
new 92a19301a16 [fix][broker]Fix deadlock when compaction and topic deletion execute concurrently (#24366)
new 639521ba516 [improve][ml]Release idle offloaded read handle only the ref count is 0 (#24381)
new 7ed6edc8edd [improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (#24386)
new b07a403f0d6 [fix][client] Fix consumer not returning encrypted messages on decryption failure with compression enabled (#24356)
new dfcc246546a [improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs (#24392)
new b55a3f7171d [improve][broker] Add managedCursor/LedgerInfoCompressionType settings to broker.conf (#24391)
new 8ae31999026 [improve][misc] Upgrade Netty to 4.1.122.Final and tcnative to 2.0.72.Final (#24397)
new 9adbbbf8f39 [fix][ml]Revert a behavior change of releasing idle offloaded ledger handle: only release idle BlobStoreBackedReadHandle (#24384)
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
buildtools/pom.xml | 2 +-
conf/broker.conf | 57 ++++++++--
conf/standalone.conf | 43 ++++++-
distribution/server/src/assemble/LICENSE.bin.txt | 54 ++++-----
distribution/shell/src/assemble/LICENSE.bin.txt | 52 ++++-----
.../bookkeeper/mledger/ManagedLedgerConfig.java | 10 ++
.../bookkeeper/mledger/OffloadedLedgerHandle.java | 4 +
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 13 ++-
pom.xml | 2 +-
.../apache/pulsar/broker/ServiceConfiguration.java | 24 +++-
.../pulsar/broker/service/AbstractTopic.java | 3 +
.../pulsar/broker/service/BrokerService.java | 2 +
...onPersistentDispatcherSingleActiveConsumer.java | 2 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 2 +-
.../broker/service/persistent/PersistentTopic.java | 1 +
.../org/apache/pulsar/client/api/RawReader.java | 11 +-
.../apache/pulsar/client/impl/RawReaderImpl.java | 23 +++-
.../org/apache/pulsar/compaction/Compactor.java | 2 +-
.../client/api/SimpleProducerConsumerTest.java | 89 +++++++++++++++
.../pulsar/client/impl/ConsumerCloseTest.java | 42 ++++++-
.../apache/pulsar/client/impl/RawReaderTest.java | 66 ++++++++++-
.../apache/pulsar/compaction/CompactionTest.java | 67 +++++++++++
.../pulsar/client/api/PulsarClientException.java | 11 ++
.../org/apache/pulsar/client/impl/ClientCnx.java | 2 +-
.../pulsar/client/impl/ConnectionHandler.java | 22 +++-
.../apache/pulsar/client/impl/ConsumerImpl.java | 126 ++++++++++++++++-----
.../apache/pulsar/client/impl/ProducerImpl.java | 39 +++++--
.../pulsar/client/impl/TopicListWatcher.java | 4 +-
.../client/impl/TransactionMetaStoreHandler.java | 4 +-
.../jcloud/impl/BlobStoreBackedReadHandleImpl.java | 24 +++-
30 files changed, 662 insertions(+), 141 deletions(-)
(pulsar) 06/08: [improve][broker] Add managedCursor/LedgerInfoCompressionType settings to broker.conf (#24391)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b55a3f7171dca2a74a75ded954e568ec8ca20848
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 9 10:20:27 2025 +0300
[improve][broker] Add managedCursor/LedgerInfoCompressionType settings to broker.conf (#24391)
(cherry picked from commit 1cc2402e07d225276f81d891bd466e7d6f11cbb9)
---
conf/broker.conf | 23 +++++++++++++++++++++++
conf/standalone.conf | 23 +++++++++++++++++++++++
2 files changed, 46 insertions(+)
diff --git a/conf/broker.conf b/conf/broker.conf
index 2a7e9ec1740..771e213736f 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1315,6 +1315,29 @@ managedLedgerUnackedRangesOpenCacheSetEnabled=true
# MetadataStore.
managedLedgerMaxUnackedRangesToPersistInMetadataStore=1000
+# ManagedCursorInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is NONE, then save the ManagedCursorInfo bytes data directly without compression.
+# Using compression reduces the size of persistent cursor (subscription) metadata. This enables using a higher
+# managedLedgerMaxUnackedRangesToPersistInMetadataStore value and reduces the overall metadata stored in
+# the metadata store such as ZooKeeper.
+managedCursorInfoCompressionType=NONE
+
+# ManagedCursorInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedCursorInfoCompressionThresholdInBytes=16384
+
+# ManagedLedgerInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is invalid or NONE, then save the ManagedLedgerInfo bytes data directly without compression.
+# Using compression reduces the size of the persistent topic metadata. When a topic contains a large number of
+# individual ledgers in BookKeeper or tiered storage, compression helps prevent the metadata size from exceeding
+# the maximum size of a metadata store entry (ZNode in ZooKeeper). This also reduces the overall metadata stored
+# in the metadata store such as ZooKeeper.
+managedLedgerInfoCompressionType=NONE
+
+# ManagedLedgerInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedLedgerInfoCompressionThresholdInBytes=16384
+
# Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
# corrupted at bookkeeper and managed-cursor is stuck at that ledger.
autoSkipNonRecoverableData=false
diff --git a/conf/standalone.conf b/conf/standalone.conf
index a23e3ebb3fc..1320b7fd482 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -849,6 +849,29 @@ managedLedgerUnackedRangesOpenCacheSetEnabled=true
# MetadataStore.
managedLedgerMaxUnackedRangesToPersistInMetadataStore=1000
+# ManagedCursorInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is NONE, then save the ManagedCursorInfo bytes data directly without compression.
+# Using compression reduces the size of persistent cursor (subscription) metadata. This enables using a higher
+# managedLedgerMaxUnackedRangesToPersistInMetadataStore value and reduces the overall metadata stored in
+# the metadata store such as ZooKeeper.
+managedCursorInfoCompressionType=NONE
+
+# ManagedCursorInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedCursorInfoCompressionThresholdInBytes=16384
+
+# ManagedLedgerInfo compression type, option values (NONE, LZ4, ZLIB, ZSTD, SNAPPY).
+# If value is invalid or NONE, then save the ManagedLedgerInfo bytes data directly without compression.
+# Using compression reduces the size of the persistent topic metadata. When a topic contains a large number of
+# individual ledgers in BookKeeper or tiered storage, compression helps prevent the metadata size from exceeding
+# the maximum size of a metadata store entry (ZNode in ZooKeeper). This also reduces the overall metadata stored
+# in the metadata store such as ZooKeeper.
+managedLedgerInfoCompressionType=NONE
+
+# ManagedLedgerInfo compression size threshold (bytes), only compress metadata when origin size more then this value.
+# 0 means compression will always apply.
+managedLedgerInfoCompressionThresholdInBytes=16384
+
# Skip reading non-recoverable/unreadable data-ledger under managed-ledger's list. It helps when data-ledgers gets
# corrupted at bookkeeper and managed-cursor is stuck at that ledger.
autoSkipNonRecoverableData=false
(pulsar) 02/08: [improve][ml]Release idle offloaded read handle only the ref count is 0 (#24381)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 639521ba5160b3191ac0fb7c10821aba54b87aac
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Jun 5 17:48:22 2025 +0800
[improve][ml]Release idle offloaded read handle only the ref count is 0 (#24381)
(cherry picked from commit 21406b594a05a6362afdf68bbe5d041a5907394d)
---
.../bookkeeper/mledger/OffloadedLedgerHandle.java | 4 ++++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 12 +++++++----
.../jcloud/impl/BlobStoreBackedReadHandleImpl.java | 24 +++++++++++++++++-----
3 files changed, 31 insertions(+), 9 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java
index f45d115090f..64464187895 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerHandle.java
@@ -26,4 +26,8 @@ public interface OffloadedLedgerHandle {
default long lastAccessTimestamp() {
return -1;
}
+
+ default int getPendingRead() {
+ return 0;
+ }
}
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 cdc7940e2c4..ede21e0beea 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
@@ -2661,10 +2661,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
ledgerCache.forEach((ledgerId, ledger) -> {
if (ledger.isDone() && !ledger.isCompletedExceptionally()) {
ReadHandle readHandle = ledger.join();
- if (readHandle instanceof OffloadedLedgerHandle) {
- long lastAccessTimestamp = ((OffloadedLedgerHandle) readHandle).lastAccessTimestamp();
- if (lastAccessTimestamp >= 0) {
- long delta = now - lastAccessTimestamp;
+ if (readHandle instanceof OffloadedLedgerHandle offloadedLedgerHandle) {
+ int pendingRead = offloadedLedgerHandle.getPendingRead();
+ if (pendingRead == 0) {
+ long delta = now - offloadedLedgerHandle.lastAccessTimestamp();
if (delta >= inactiveOffloadedLedgerEvictionTimeMs) {
log.info("[{}] Offloaded ledger {} can be released ({} ms elapsed since last access)",
name, ledgerId, delta);
@@ -2674,6 +2674,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
"[{}] Offloaded ledger {} cannot be released ({} ms elapsed since last access)",
name, ledgerId, delta);
}
+ } else if (pendingRead < 0) {
+ log.error("[{}] Offloaded ledger {} went to a wrong state because its pending read is a"
+ + " negative value {}. Please raise an issue to https://github.com/apache/pulsar", name,
+ ledgerId, pendingRead);
}
}
}
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
index 1f2f901f514..84d9d118728 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
@@ -29,6 +29,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
@@ -56,6 +57,9 @@ import org.slf4j.LoggerFactory;
public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedgerHandle {
private static final Logger log = LoggerFactory.getLogger(BlobStoreBackedReadHandleImpl.class);
+ protected static final AtomicIntegerFieldUpdater<BlobStoreBackedReadHandleImpl> PENDING_READ_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(BlobStoreBackedReadHandleImpl.class, "pendingRead");
+
private final long ledgerId;
private final OffloadIndexBlock index;
private final BackedInputStream inputStream;
@@ -71,6 +75,8 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
private volatile State state = null;
+ private volatile int pendingRead;
+
private volatile long lastAccessTimestamp = System.currentTimeMillis();
private BlobStoreBackedReadHandleImpl(long ledgerId, OffloadIndexBlock index,
@@ -122,9 +128,17 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
getId(), firstEntry, lastEntry, (1 + lastEntry - firstEntry));
}
CompletableFuture<LedgerEntries> promise = new CompletableFuture<>();
- touch();
+
+ // Ledger handles will be only marked idle when "pendingRead" is "0", it is not needed to update
+ // "lastAccessTimestamp" if "pendingRead" is larger than "0".
+ // Rather than update "lastAccessTimestamp" when starts a reading, updating it when a reading task is finished
+ // is better.
+ PENDING_READ_UPDATER.incrementAndGet(this);
+ promise.whenComplete((__, ex) -> {
+ PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
+ lastAccessTimestamp = System.currentTimeMillis();
+ });
executor.execute(() -> {
- touch();
if (state == State.Closed) {
log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}",
ledgerId, firstEntry, lastEntry);
@@ -213,7 +227,6 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
}
private void seekToEntry(long nextExpectedId) throws IOException {
- touch();
Long knownOffset = entryOffsetsCache.getIfPresent(ledgerId, nextExpectedId);
if (knownOffset != null) {
inputStream.seek(knownOffset);
@@ -324,7 +337,8 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
return lastAccessTimestamp;
}
- private void touch() {
- lastAccessTimestamp = System.currentTimeMillis();
+ @Override
+ public int getPendingRead() {
+ return PENDING_READ_UPDATER.get(this);
}
}
(pulsar) 05/08: [improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs (#24392)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dfcc246546a160d3abd5d7e46a96d53aae9b48b7
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 9 10:20:14 2025 +0300
[improve][broker] Make maxBatchDeletedIndexToPersist configurable and document other related configs (#24392)
(cherry picked from commit d1bca65f9b8907308efe69c887516d3c6b514a10)
---
conf/broker.conf | 34 +++++++++++++++-------
conf/standalone.conf | 20 +++++++++++--
.../bookkeeper/mledger/ManagedLedgerConfig.java | 10 +++++++
.../apache/pulsar/broker/ServiceConfiguration.java | 24 +++++++++++----
.../pulsar/broker/service/BrokerService.java | 2 ++
5 files changed, 72 insertions(+), 18 deletions(-)
diff --git a/conf/broker.conf b/conf/broker.conf
index 59bc1d6820b..2a7e9ec1740 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -1276,6 +1276,14 @@ managedLedgerCursorMaxEntriesPerLedger=50000
# Max time before triggering a rollover on a cursor ledger
managedLedgerCursorRolloverTimeInSeconds=14400
+# Maximum amount of memory used hold data read from storage (or from the cache).
+# This mechanism prevents the broker to have too many concurrent
+# reads from storage and fall into Out of Memory errors in case
+# of multiple concurrent reads to multiple concurrent consumers.
+# Set 0 in order to disable the feature.
+#
+managedLedgerMaxReadsInFlightSizeInMB=0
+
# Max number of "acknowledgment holes" that are going to be persistently stored.
# When acknowledging out of order, a consumer will leave holes that are supposed
# to be quickly filled by acking all the messages. The information of which
@@ -1285,13 +1293,22 @@ managedLedgerCursorRolloverTimeInSeconds=14400
# crashes.
managedLedgerMaxUnackedRangesToPersist=10000
-# Maximum amount of memory used hold data read from storage (or from the cache).
-# This mechanism prevents the broker to have too many concurrent
-# reads from storage and fall into Out of Memory errors in case
-# of multiple concurrent reads to multiple concurrent consumers.
-# Set 0 in order to disable the feature.
-#
-managedLedgerMaxReadsInFlightSizeInMB=0
+# Maximum number of partially acknowledged batch messages per subscription that will have their batch
+# deleted indexes persisted. Batch deleted index state is handled when acknowledgmentAtBatchIndexLevelEnabled=true.
+# When this limit is exceeded, remaining batch message containing the batch deleted indexes will
+# only be tracked in memory. In case of broker restarts or load balancing events, the batch
+# deleted indexes will be cleared while redelivering the messages to consumers.
+managedLedgerMaxBatchDeletedIndexToPersist=10000
+
+# When storing acknowledgement state, choose a more compact serialization format that stores
+# individual acknowledgements as a bitmap which is serialized to an array of long values. NOTE: This setting requires
+# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+managedLedgerPersistIndividualAckAsLongArray=false
+
+# When set to true, a BitSet will be used to track acknowledged messages that come after the "mark delete position"
+# for each subscription. RoaringBitmap is used as a memory efficient BitSet implementation for the acknowledged
+# messages tracking. Unacknowledged ranges are the message ranges excluding the acknowledged messages.
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
# Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
@@ -1770,9 +1787,6 @@ narExtractionDirectory=
# Maximum prefetch rounds for ledger reading for offloading
managedLedgerOffloadPrefetchRounds=1
-# Use Open Range-Set to cache unacked messages
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
# For Amazon S3 ledger offload, AWS region
s3ManagedLedgerOffloadRegion=
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 33cbebbbc9f..a23e3ebb3fc 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -827,6 +827,23 @@ managedLedgerMaxSizePerLedgerMbytes=2048
# crashes.
managedLedgerMaxUnackedRangesToPersist=10000
+# Maximum number of partially acknowledged batch messages per subscription that will have their batch
+# deleted indexes persisted. Batch deleted index state is handled when acknowledgmentAtBatchIndexLevelEnabled=true.
+# When this limit is exceeded, remaining batch message containing the batch deleted indexes will
+# only be tracked in memory. In case of broker restarts or load balancing events, the batch
+# deleted indexes will be cleared while redelivering the messages to consumers.
+managedLedgerMaxBatchDeletedIndexToPersist=10000
+
+# When storing acknowledgement state, choose a more compact serialization format that stores
+# individual acknowledgements as a bitmap which is serialized to an array of long values. NOTE: This setting requires
+# managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.
+managedLedgerPersistIndividualAckAsLongArray=false
+
+# When set to true, a BitSet will be used to track acknowledged messages that come after the "mark delete position"
+# for each subscription. RoaringBitmap is used as a memory efficient BitSet implementation for the acknowledged
+# messages tracking. Unacknowledged ranges are the message ranges excluding the acknowledged messages.
+managedLedgerUnackedRangesOpenCacheSetEnabled=true
+
# Max number of "acknowledgment holes" that can be stored in MetadataStore. If number of unack message range is higher
# than this limit then broker will persist unacked ranges into bookkeeper to avoid additional data overhead into
# MetadataStore.
@@ -861,9 +878,6 @@ managedLedgerMinimumBacklogEntriesForCaching=1000
# Maximum backlog entry difference to prevent caching entries that can't be reused.
managedLedgerMaxBacklogBetweenCursorsForCaching=1000
-# Use Open Range-Set to cache unacked messages
-managedLedgerUnackedRangesOpenCacheSetEnabled=true
-
# Managed ledger prometheus stats latency rollover seconds (default: 60s)
managedLedgerPrometheusStatsLatencyRolloverSeconds=60
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 c1cbf0ad704..aaa127973a5 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
@@ -502,6 +502,16 @@ public class ManagedLedgerConfig {
return maxBatchDeletedIndexToPersist;
}
+ /**
+ * Set max batch deleted index that will be persisted and recovered.
+ *
+ * @param maxBatchDeletedIndexToPersist
+ * max batch deleted index that will be persisted and recovered.
+ */
+ public void setMaxBatchDeletedIndexToPersist(int maxBatchDeletedIndexToPersist) {
+ this.maxBatchDeletedIndexToPersist = maxBatchDeletedIndexToPersist;
+ }
+
public boolean isPersistentUnackedRangesWithMultipleEntriesEnabled() {
return persistentUnackedRangesWithMultipleEntriesEnabled;
}
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 1d023830413..1a6f330a43e 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
@@ -2265,10 +2265,22 @@ public class ServiceConfiguration implements PulsarConfiguration {
+ " will only be tracked in memory and messages will be redelivered in case of"
+ " crashes.")
private int managedLedgerMaxUnackedRangesToPersist = 10000;
- @FieldContext(
- category = CATEGORY_STORAGE_ML,
- doc = "Whether persist cursor ack stats as long arrays, which will compress the data and reduce GC rate")
+
+ @FieldContext(category = CATEGORY_STORAGE_ML,
+ doc = "Maximum number of partially acknowledged batch messages per subscription that will have their batch "
+ + "deleted indexes persisted. Batch deleted index state is handled when "
+ + "acknowledgmentAtBatchIndexLevelEnabled=true.\n\n"
+ + "When this limit is exceeded, remaining batch message containing the batch deleted indexes will "
+ + "only be tracked in memory. In case of broker restarts or load balancing events, the batch "
+ + "deleted indexes will be cleared while redelivering the messages to consumers.")
+ private int managedLedgerMaxBatchDeletedIndexToPersist = 10000;
+
+ @FieldContext(category = CATEGORY_STORAGE_ML,
+ doc = "When storing acknowledgement state, choose a more compact serialization format that stores"
+ + " individual acknowledgements as a bitmap which is serialized to an array of long values.\n\n"
+ + "NOTE: This setting requires managedLedgerUnackedRangesOpenCacheSetEnabled=true to be effective.")
private boolean managedLedgerPersistIndividualAckAsLongArray = false;
+
@FieldContext(
category = CATEGORY_STORAGE_ML,
doc = "If enabled, the maximum \"acknowledgment holes\" will not be limited and \"acknowledgment holes\" "
@@ -2291,8 +2303,10 @@ public class ServiceConfiguration implements PulsarConfiguration {
private int managedLedgerMaxUnackedRangesToPersistInMetadataStore = 1000;
@FieldContext(
category = CATEGORY_STORAGE_OFFLOADING,
- doc = "Use Open Range-Set to cache unacked messages (it is memory efficient but it can take more cpu)"
- )
+ doc = "When set to true, a BitSet will be used to track acknowledged messages that come after the \"mark "
+ + "delete position\" for each subscription.\n\nRoaringBitmap is used as a memory efficient BitSet "
+ + "implementation for the acknowledged messages tracking. Unacknowledged ranges are the message "
+ + "ranges excluding the acknowledged messages.")
private boolean managedLedgerUnackedRangesOpenCacheSetEnabled = true;
@FieldContext(
dynamic = true,
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 769ac66e0ca..24d89ff2873 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
@@ -2044,6 +2044,8 @@ public class BrokerService implements Closeable {
managedLedgerConfig
.setMaxUnackedRangesToPersist(serviceConfig.getManagedLedgerMaxUnackedRangesToPersist());
+ managedLedgerConfig.setMaxBatchDeletedIndexToPersist(
+ serviceConfig.getManagedLedgerMaxBatchDeletedIndexToPersist());
managedLedgerConfig
.setPersistIndividualAckAsLongArray(serviceConfig.isManagedLedgerPersistIndividualAckAsLongArray());
managedLedgerConfig.setPersistentUnackedRangesWithMultipleEntriesEnabled(
(pulsar) 07/08: [improve][misc] Upgrade Netty to 4.1.122.Final and tcnative to 2.0.72.Final (#24397)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8ae3199902685ea4ea59b669eba9aa6492ae1093
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Jun 9 11:56:01 2025 +0300
[improve][misc] Upgrade Netty to 4.1.122.Final and tcnative to 2.0.72.Final (#24397)
(cherry picked from commit db459477dd2548bebbbd519af8ade9e7857fa0a8)
---
buildtools/pom.xml | 2 +-
distribution/server/src/assemble/LICENSE.bin.txt | 54 ++++++++++++------------
distribution/shell/src/assemble/LICENSE.bin.txt | 52 +++++++++++------------
pom.xml | 2 +-
4 files changed, 55 insertions(+), 55 deletions(-)
diff --git a/buildtools/pom.xml b/buildtools/pom.xml
index 748f69a28b0..4815cadfa4e 100644
--- a/buildtools/pom.xml
+++ b/buildtools/pom.xml
@@ -55,7 +55,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.121.Final</netty.version>
+ <netty.version>4.1.122.Final</netty.version>
<guice.version>4.2.3</guice.version>
<guava.version>32.1.2-jre</guava.version>
<ant.version>1.10.12</ant.version>
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index f3ce3ba8374..f2f2f52bf41 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -293,33 +293,33 @@ The Apache Software License, Version 2.0
- org.apache.commons-commons-lang3-3.17.0.jar
- org.apache.commons-commons-text-1.13.1.jar
* Netty
- - io.netty-netty-buffer-4.1.121.Final.jar
- - io.netty-netty-codec-4.1.121.Final.jar
- - io.netty-netty-codec-dns-4.1.121.Final.jar
- - io.netty-netty-codec-http-4.1.121.Final.jar
- - io.netty-netty-codec-http2-4.1.121.Final.jar
- - io.netty-netty-codec-socks-4.1.121.Final.jar
- - io.netty-netty-codec-haproxy-4.1.121.Final.jar
- - io.netty-netty-common-4.1.121.Final.jar
- - io.netty-netty-handler-4.1.121.Final.jar
- - io.netty-netty-handler-proxy-4.1.121.Final.jar
- - io.netty-netty-resolver-4.1.121.Final.jar
- - io.netty-netty-resolver-dns-4.1.121.Final.jar
- - io.netty-netty-resolver-dns-classes-macos-4.1.121.Final.jar
- - io.netty-netty-resolver-dns-native-macos-4.1.121.Final-osx-aarch_64.jar
- - io.netty-netty-resolver-dns-native-macos-4.1.121.Final-osx-x86_64.jar
- - io.netty-netty-transport-4.1.121.Final.jar
- - io.netty-netty-transport-classes-epoll-4.1.121.Final.jar
- - io.netty-netty-transport-native-epoll-4.1.121.Final-linux-aarch_64.jar
- - io.netty-netty-transport-native-epoll-4.1.121.Final-linux-x86_64.jar
- - io.netty-netty-transport-native-unix-common-4.1.121.Final.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-linux-aarch_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-linux-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-osx-aarch_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-osx-x86_64.jar
- - io.netty-netty-tcnative-boringssl-static-2.0.70.Final-windows-x86_64.jar
- - io.netty-netty-tcnative-classes-2.0.70.Final.jar
+ - io.netty-netty-buffer-4.1.122.Final.jar
+ - io.netty-netty-codec-4.1.122.Final.jar
+ - io.netty-netty-codec-dns-4.1.122.Final.jar
+ - io.netty-netty-codec-http-4.1.122.Final.jar
+ - io.netty-netty-codec-http2-4.1.122.Final.jar
+ - io.netty-netty-codec-socks-4.1.122.Final.jar
+ - io.netty-netty-codec-haproxy-4.1.122.Final.jar
+ - io.netty-netty-common-4.1.122.Final.jar
+ - io.netty-netty-handler-4.1.122.Final.jar
+ - io.netty-netty-handler-proxy-4.1.122.Final.jar
+ - io.netty-netty-resolver-4.1.122.Final.jar
+ - io.netty-netty-resolver-dns-4.1.122.Final.jar
+ - io.netty-netty-resolver-dns-classes-macos-4.1.122.Final.jar
+ - io.netty-netty-resolver-dns-native-macos-4.1.122.Final-osx-aarch_64.jar
+ - io.netty-netty-resolver-dns-native-macos-4.1.122.Final-osx-x86_64.jar
+ - io.netty-netty-transport-4.1.122.Final.jar
+ - io.netty-netty-transport-classes-epoll-4.1.122.Final.jar
+ - io.netty-netty-transport-native-epoll-4.1.122.Final-linux-aarch_64.jar
+ - io.netty-netty-transport-native-epoll-4.1.122.Final-linux-x86_64.jar
+ - io.netty-netty-transport-native-unix-common-4.1.122.Final.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.72.Final.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-linux-aarch_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-linux-x86_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-osx-aarch_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-osx-x86_64.jar
+ - io.netty-netty-tcnative-boringssl-static-2.0.72.Final-windows-x86_64.jar
+ - io.netty-netty-tcnative-classes-2.0.72.Final.jar
- io.netty.incubator-netty-incubator-transport-classes-io_uring-0.0.26.Final.jar
- io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.26.Final-linux-x86_64.jar
- io.netty.incubator-netty-incubator-transport-native-io_uring-0.0.26.Final-linux-aarch_64.jar
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index e77a0f1c459..6bd5fca8cad 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -349,35 +349,35 @@ The Apache Software License, Version 2.0
- commons-text-1.13.1.jar
- commons-compress-1.27.1.jar
* Netty
- - netty-buffer-4.1.121.Final.jar
- - netty-codec-4.1.121.Final.jar
- - netty-codec-dns-4.1.121.Final.jar
- - netty-codec-http-4.1.121.Final.jar
- - netty-codec-socks-4.1.121.Final.jar
- - netty-codec-haproxy-4.1.121.Final.jar
- - netty-common-4.1.121.Final.jar
- - netty-handler-4.1.121.Final.jar
- - netty-handler-proxy-4.1.121.Final.jar
- - netty-resolver-4.1.121.Final.jar
- - netty-resolver-dns-4.1.121.Final.jar
- - netty-transport-4.1.121.Final.jar
- - netty-transport-classes-epoll-4.1.121.Final.jar
- - netty-transport-native-epoll-4.1.121.Final-linux-aarch_64.jar
- - netty-transport-native-epoll-4.1.121.Final-linux-x86_64.jar
- - netty-transport-native-unix-common-4.1.121.Final.jar
- - netty-tcnative-boringssl-static-2.0.70.Final.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-linux-aarch_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-linux-x86_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-osx-aarch_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-osx-x86_64.jar
- - netty-tcnative-boringssl-static-2.0.70.Final-windows-x86_64.jar
- - netty-tcnative-classes-2.0.70.Final.jar
+ - netty-buffer-4.1.122.Final.jar
+ - netty-codec-4.1.122.Final.jar
+ - netty-codec-dns-4.1.122.Final.jar
+ - netty-codec-http-4.1.122.Final.jar
+ - netty-codec-socks-4.1.122.Final.jar
+ - netty-codec-haproxy-4.1.122.Final.jar
+ - netty-common-4.1.122.Final.jar
+ - netty-handler-4.1.122.Final.jar
+ - netty-handler-proxy-4.1.122.Final.jar
+ - netty-resolver-4.1.122.Final.jar
+ - netty-resolver-dns-4.1.122.Final.jar
+ - netty-transport-4.1.122.Final.jar
+ - netty-transport-classes-epoll-4.1.122.Final.jar
+ - netty-transport-native-epoll-4.1.122.Final-linux-aarch_64.jar
+ - netty-transport-native-epoll-4.1.122.Final-linux-x86_64.jar
+ - netty-transport-native-unix-common-4.1.122.Final.jar
+ - netty-tcnative-boringssl-static-2.0.72.Final.jar
+ - netty-tcnative-boringssl-static-2.0.72.Final-linux-aarch_64.jar
+ - netty-tcnative-boringssl-static-2.0.72.Final-linux-x86_64.jar
+ - netty-tcnative-boringssl-static-2.0.72.Final-osx-aarch_64.jar
+ - netty-tcnative-boringssl-static-2.0.72.Final-osx-x86_64.jar
+ - netty-tcnative-boringssl-static-2.0.72.Final-windows-x86_64.jar
+ - netty-tcnative-classes-2.0.72.Final.jar
- netty-incubator-transport-classes-io_uring-0.0.26.Final.jar
- netty-incubator-transport-native-io_uring-0.0.26.Final-linux-aarch_64.jar
- netty-incubator-transport-native-io_uring-0.0.26.Final-linux-x86_64.jar
- - netty-resolver-dns-classes-macos-4.1.121.Final.jar
- - netty-resolver-dns-native-macos-4.1.121.Final-osx-aarch_64.jar
- - netty-resolver-dns-native-macos-4.1.121.Final-osx-x86_64.jar
+ - netty-resolver-dns-classes-macos-4.1.122.Final.jar
+ - netty-resolver-dns-native-macos-4.1.122.Final-osx-aarch_64.jar
+ - netty-resolver-dns-native-macos-4.1.122.Final-osx-x86_64.jar
* Prometheus client
- simpleclient-0.16.0.jar
- simpleclient_log4j2-0.16.0.jar
diff --git a/pom.xml b/pom.xml
index 541c1029913..98eecb31abf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -184,7 +184,7 @@ flexible messaging model and an intuitive client API.</description>
<snappy.version>1.1.10.5</snappy.version> <!-- ZooKeeper server -->
<dropwizardmetrics.version>4.1.12.1</dropwizardmetrics.version> <!-- ZooKeeper server -->
<curator.version>5.7.1</curator.version>
- <netty.version>4.1.121.Final</netty.version>
+ <netty.version>4.1.122.Final</netty.version>
<netty-iouring.version>0.0.26.Final</netty-iouring.version>
<jetty.version>9.4.57.v20241219</jetty.version>
<conscrypt.version>2.5.2</conscrypt.version>
(pulsar) 04/08: [fix][client] Fix consumer not returning encrypted messages on decryption failure with compression enabled (#24356)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b07a403f0d62039b9c2a2a5a77198e9372dd4291
Author: Zike Yang <zi...@apache.org>
AuthorDate: Fri Jun 6 18:36:55 2025 +0800
[fix][client] Fix consumer not returning encrypted messages on decryption failure with compression enabled (#24356)
## Motivation
There is a problem when a consumer tries to decrypt a compressed message. If a consumer fails to decrypt a compressed message, it also fails to decompress it, leading to the message being discarded. The user won't receive the message, even if using `CryptoFailureAction.CONSUME`, because the failure occurs during decompression, not decryption. This issue could be easily reproduced by the test added in this PR: `testE2EEncryptionWithCompression `.
We already have logic to skip decompression for messages that can't be decrypted: https://github.com/apache/pulsar/blob/397b0211cf1c9069d80daa691ca530e09bc37999/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1439-L1441
The main issue is if the consumer fails to decrypt the message here: https://github.com/apache/pulsar/blob/397b0211cf1c9069d80daa691ca530e09bc37999/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1947, it isn't treated as undecryptable and still tries to decompress.
## Modification
- Refactor `decryptPayloadIfNeeded` to return a clearer result, `DecryptResult`, to guide the logic on whether the message was decrypted successfully or should be discarded.
- If decryption fails, treat the message as undecryptable and skip decompression.
(cherry picked from commit 9c954548b6636a3d5ef14eef05338c881daaa80d)
---
.../client/api/SimpleProducerConsumerTest.java | 89 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 70 +++++++++++++----
2 files changed, 146 insertions(+), 13 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 9e35b4f262e..ddb418c37a7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -5062,4 +5062,93 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
.createAsync().get(5, TimeUnit.SECONDS);
assertNotNull(reader);
}
+
+ /**
+ * Creates a CryptoKeyReader that always returns keys from specified file paths,
+ * regardless of the key name provided.
+ *
+ * <p>This method creates a CryptoKeyReader instance that reads the public and
+ * private keys from fixed file paths. The key name and metadata provided to
+ * the getPublicKey and getPrivateKey methods are ignored, and the keys are
+ * always read from the specified paths.</p>
+ *
+ * @param publicKeyPath the file path to the public key
+ * @param privateKeyPath the file path to the private key
+ * @return a CryptoKeyReader that reads keys from the specified file paths
+ * @throws AssertionError if the key files are not present or not readable
+ */
+ private static CryptoKeyReader createFixedFileCryptoKeyReader(String publicKeyPath, String privateKeyPath) {
+ return new CryptoKeyReader() {
+ final EncryptionKeyInfo keyInfo = new EncryptionKeyInfo();
+
+ @Override
+ public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> keyMeta) {
+ if (Files.isReadable(Paths.get(publicKeyPath))) {
+ try {
+ keyInfo.setKey(Files.readAllBytes(Paths.get(publicKeyPath)));
+ return keyInfo;
+ } catch (IOException e) {
+ Assert.fail("Failed to read certificate from " + publicKeyPath);
+ }
+ } else {
+ Assert.fail("Certificate file " + publicKeyPath + " is not present or not readable.");
+ }
+ return null;
+ }
+
+ @Override
+ public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> keyMeta) {
+ if (Files.isReadable(Paths.get(privateKeyPath))) {
+ try {
+ keyInfo.setKey(Files.readAllBytes(Paths.get(privateKeyPath)));
+ return keyInfo;
+ } catch (IOException e) {
+ Assert.fail("Failed to read certificate from " + privateKeyPath);
+ }
+ } else {
+ Assert.fail("Certificate file " + privateKeyPath + " is not present or not readable.");
+ }
+ return null;
+ }
+ };
+ }
+
+ @Test
+ public void testE2EEncryptionWithCompression() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testE2EEncryptionWithCompression-" + UUID.randomUUID();
+
+ final var producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .addEncryptionKey("client-rsa.pem")
+ .cryptoKeyReader(createFixedFileCryptoKeyReader(
+ "./src/test/resources/certificate/public-key.client-rsa.pem",
+ "./src/test/resources/certificate/private-key.client-rsa.pem"
+ ))
+ //.compressionMinMsgBodySize(1) // enforce compression
+ .compressionType(CompressionType.LZ4)
+ .create();
+
+ final var consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test")
+ .cryptoKeyReader(createFixedFileCryptoKeyReader(
+ "./src/test/resources/certificate/public-key.client-mismatch-rsa.pem",
+ "./src/test/resources/certificate/private-key.client-mismatch-rsa.pem"
+ ))
+ .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME)
+ .subscribe();
+
+ for (int i = 0; i < 10; i++) {
+ producer.send("message-" + i);
+ }
+
+ for (int i = 0; i < 10; i++) {
+ final var msg = consumer.receive(5, TimeUnit.SECONDS);
+ assertNotNull(msg);
+ consumer.acknowledge(msg);
+ }
+
+ producer.close();
+ consumer.close();
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index f575a127c96..b4a9ec610e8 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1456,16 +1456,18 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return;
}
- ByteBuf decryptedPayload = decryptPayloadIfNeeded(messageId, redeliveryCount, msgMetadata, headersAndPayload,
+ DecryptResult decryptResult = decryptPayloadIfNeeded(messageId, redeliveryCount, msgMetadata, headersAndPayload,
cnx);
- boolean isMessageUndecryptable = isMessageUndecryptable(msgMetadata);
-
- if (decryptedPayload == null) {
+ if (decryptResult.shouldDiscard()) {
// Message was discarded or CryptoKeyReader isn't implemented
return;
}
+ boolean isMessageUndecryptable = !decryptResult.success;
+
+ ByteBuf decryptedPayload = decryptResult.payload;
+
// uncompress decryptedPayload and release decryptedPayload-ByteBuf
ByteBuf uncompressedPayload = (isMessageUndecryptable || isChunkedMessage) ? decryptedPayload.retain()
: uncompressPayloadIfNeeded(messageId, msgMetadata, decryptedPayload, cnx, true);
@@ -1951,11 +1953,53 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return connectionHandler.lastConnectionClosedTimestamp;
}
- private ByteBuf decryptPayloadIfNeeded(MessageIdData messageId, int redeliveryCount, MessageMetadata msgMetadata,
- ByteBuf payload, ClientCnx currentCnx) {
+ /**
+ * Represents the outcome of a message decryption attempt for the consumer.
+ */
+ private static class DecryptResult {
+ private final boolean success;
+ private final ByteBuf payload;
+
+ private DecryptResult(boolean success, ByteBuf decryptedPayload) {
+ this.success = success;
+ this.payload = decryptedPayload;
+ }
+
+ /**
+ * Returns true if the message should be discarded and not delivered to the consumer user.
+ */
+ public boolean shouldDiscard() {
+ return this.payload == null;
+ }
+
+ /**
+ * Creates a result indicating decryption succeeded and the payload is ready for use.
+ */
+ public static DecryptResult success(ByteBuf decryptedPayload) {
+ return new DecryptResult(true, decryptedPayload);
+ }
+
+ /**
+ * Creates a result indicating decryption failed, but the message should still be delivered.
+ */
+ public static DecryptResult failure(ByteBuf decryptedPayload) {
+ return new DecryptResult(false, decryptedPayload);
+ }
+
+ /**
+ * Creates a result indicating the message should be discarded.
+ */
+ public static DecryptResult discard() {
+ return new DecryptResult(false, null);
+ }
+ }
+
+ private DecryptResult decryptPayloadIfNeeded(MessageIdData messageId, int redeliveryCount,
+ MessageMetadata msgMetadata,
+ ByteBuf payload, ClientCnx currentCnx) {
if (msgMetadata.getEncryptionKeysCount() == 0) {
- return payload.retain();
+ return DecryptResult.success(payload.retain());
}
int batchSize = msgMetadata.getNumMessagesInBatch();
// If KeyReader is not configured throw exception based on config param
@@ -1969,7 +2013,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
ByteBuffer nioDecryptedData = decryptedData.nioBuffer(0, maxDecryptedSize);
if (msgCrypto.decrypt(() -> msgMetadata, payload.nioBuffer(), nioDecryptedData, conf.getCryptoKeyReader())) {
decryptedData.writerIndex(nioDecryptedData.limit());
- return decryptedData;
+ return DecryptResult.success(decryptedData);
}
decryptedData.release();
@@ -1977,7 +2021,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return handleCryptoFailure(payload, messageId, currentCnx, redeliveryCount, batchSize, false);
}
- private ByteBuf handleCryptoFailure(ByteBuf payload, MessageIdData messageId, ClientCnx currentCnx,
+ private DecryptResult handleCryptoFailure(ByteBuf payload, MessageIdData messageId, ClientCnx currentCnx,
int redeliveryCount, int batchSize, boolean cryptoReaderNotExist) {
switch (conf.getCryptoFailureAction()) {
@@ -1990,7 +2034,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
log.warn("[{}][{}][{}][{}] Decryption failed. Consuming encrypted message since config is set to"
+ " consume.", topic, subscription, consumerName, messageId);
}
- return payload.retain();
+ return DecryptResult.failure(payload.retain());
case DISCARD:
if (cryptoReaderNotExist) {
log.warn(
@@ -2005,7 +2049,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
messageId.getBatchIndex());
}
discardMessage(messageId, currentCnx, ValidationError.DecryptionError, batchSize);
- return null;
+ return DecryptResult.discard();
case FAIL:
if (cryptoReaderNotExist) {
log.error(
@@ -2020,11 +2064,11 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
MessageId m = new MessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(), partitionIndex);
unAckedMessageTracker.add(m, redeliveryCount);
- return null;
+ return DecryptResult.discard();
default:
log.warn("[{}][{}][{}] Invalid crypto failure state found, continue message consumption.", topic,
subscription, consumerName);
- return payload.retain();
+ return DecryptResult.failure(payload.retain());
}
}
(pulsar) 01/08: [fix][broker]Fix deadlock when compaction and topic deletion execute concurrently (#24366)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 92a19301a16b6879f43168fbd6fb25666e7e5b06
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Jun 5 10:02:53 2025 +0800
[fix][broker]Fix deadlock when compaction and topic deletion execute concurrently (#24366)
(cherry picked from commit 37e160f78f1504d3cd45523a08872f8f7d461df7)
---
.../pulsar/broker/service/AbstractTopic.java | 3 +
.../broker/service/persistent/PersistentTopic.java | 1 +
.../org/apache/pulsar/client/api/RawReader.java | 11 ++--
.../apache/pulsar/client/impl/RawReaderImpl.java | 23 ++++++--
.../org/apache/pulsar/compaction/Compactor.java | 2 +-
.../pulsar/client/impl/ConsumerCloseTest.java | 42 +++++++++++++-
.../apache/pulsar/client/impl/RawReaderTest.java | 66 ++++++++++++++++++++-
.../apache/pulsar/compaction/CompactionTest.java | 67 ++++++++++++++++++++++
.../pulsar/client/api/PulsarClientException.java | 11 ++++
.../org/apache/pulsar/client/impl/ClientCnx.java | 2 +-
.../pulsar/client/impl/ConnectionHandler.java | 22 +++++--
.../apache/pulsar/client/impl/ConsumerImpl.java | 56 +++++++++++++-----
.../apache/pulsar/client/impl/ProducerImpl.java | 39 ++++++++++---
.../pulsar/client/impl/TopicListWatcher.java | 4 +-
.../client/impl/TransactionMetaStoreHandler.java | 4 +-
15 files changed, 309 insertions(+), 44 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 20b2abfa48e..6d3c9a58b03 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;
import static org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl.ENTRY_LATENCY_BUCKETS_USEC;
import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import java.time.Clock;
import java.util.ArrayList;
@@ -114,6 +115,8 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ @VisibleForTesting
+ @Getter
protected volatile boolean isFenced;
protected final HierarchyTopicPolicies topicPolicies;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index e147796d5a6..31024f6976c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -270,6 +270,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private volatile double lastUpdatedAvgPublishRateInMsg = 0;
private volatile double lastUpdatedAvgPublishRateInByte = 0;
+ @Getter
private volatile boolean isClosingOrDeleting = false;
private ScheduledFuture<?> fencedTopicMonitoringTask = null;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
index 55483708fdf..ae4927da5ce 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
@@ -33,23 +33,24 @@ public interface RawReader {
*/
static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription) {
- return create(client, topic, subscription, true);
+ return create(client, topic, subscription, true, true);
}
static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription,
- boolean createTopicIfDoesNotExist) {
+ boolean createTopicIfDoesNotExist, boolean retryOnRecoverableErrors) {
CompletableFuture<Consumer<byte[]>> future = new CompletableFuture<>();
RawReader r =
- new RawReaderImpl((PulsarClientImpl) client, topic, subscription, future, createTopicIfDoesNotExist);
+ new RawReaderImpl((PulsarClientImpl) client, topic, subscription, future, createTopicIfDoesNotExist,
+ retryOnRecoverableErrors);
return future.thenApply(__ -> r);
}
static CompletableFuture<RawReader> create(PulsarClient client,
ConsumerConfigurationData<byte[]> consumerConfiguration,
- boolean createTopicIfDoesNotExist) {
+ boolean createTopicIfDoesNotExist, boolean retryOnRecoverableErrors) {
CompletableFuture<Consumer<byte[]>> future = new CompletableFuture<>();
RawReader r = new RawReaderImpl((PulsarClientImpl) client,
- consumerConfiguration, future, createTopicIfDoesNotExist);
+ consumerConfiguration, future, createTopicIfDoesNotExist, retryOnRecoverableErrors);
return future.thenApply(__ -> r);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
index 5ac051d2271..32f75d71dc3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -28,6 +28,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.api.RawReader;
import org.apache.pulsar.client.api.Schema;
@@ -52,7 +53,7 @@ public class RawReaderImpl implements RawReader {
public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
CompletableFuture<Consumer<byte[]>> consumerFuture,
- boolean createTopicIfDoesNotExist) {
+ boolean createTopicIfDoesNotExist, boolean retryOnRecoverableErrors) {
consumerConfiguration = new ConsumerConfigurationData<>();
consumerConfiguration.getTopicNames().add(topic);
consumerConfiguration.setSubscriptionName(subscription);
@@ -62,14 +63,16 @@ public class RawReaderImpl implements RawReader {
consumerConfiguration.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
consumerConfiguration.setAckReceiptEnabled(true);
- consumer = new RawConsumerImpl(client, consumerConfiguration, consumerFuture, createTopicIfDoesNotExist);
+ consumer = new RawConsumerImpl(client, consumerConfiguration, consumerFuture, createTopicIfDoesNotExist,
+ retryOnRecoverableErrors);
}
public RawReaderImpl(PulsarClientImpl client, ConsumerConfigurationData<byte[]> consumerConfiguration,
CompletableFuture<Consumer<byte[]>> consumerFuture,
- boolean createTopicIfDoesNotExist) {
+ boolean createTopicIfDoesNotExist, boolean retryOnRecoverableErrors) {
this.consumerConfiguration = consumerConfiguration;
- consumer = new RawConsumerImpl(client, consumerConfiguration, consumerFuture, createTopicIfDoesNotExist);
+ consumer = new RawConsumerImpl(client, consumerConfiguration, consumerFuture, createTopicIfDoesNotExist,
+ retryOnRecoverableErrors);
}
@@ -117,9 +120,11 @@ public class RawReaderImpl implements RawReader {
static class RawConsumerImpl extends ConsumerImpl<byte[]> {
final BlockingQueue<RawMessageAndCnx> incomingRawMessages;
final Queue<CompletableFuture<RawMessage>> pendingRawReceives;
+ final boolean retryOnRecoverableErrors;
RawConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<byte[]> conf,
- CompletableFuture<Consumer<byte[]>> consumerFuture, boolean createTopicIfDoesNotExist) {
+ CompletableFuture<Consumer<byte[]>> consumerFuture, boolean createTopicIfDoesNotExist,
+ boolean retryOnRecoverableErrors) {
super(client,
conf.getSingleTopic(),
conf,
@@ -135,6 +140,14 @@ public class RawReaderImpl implements RawReader {
);
incomingRawMessages = new GrowableArrayBlockingQueue<>();
pendingRawReceives = new ConcurrentLinkedQueue<>();
+ this.retryOnRecoverableErrors = retryOnRecoverableErrors;
+ }
+
+ protected boolean isUnrecoverableError(Throwable t) {
+ if (!retryOnRecoverableErrors && (t instanceof PulsarClientException.ServiceNotReadyException)) {
+ return true;
+ }
+ return super.isUnrecoverableError(t);
}
void tryCompletePending() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java
index 983443432ff..d37298757db 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/Compactor.java
@@ -56,7 +56,7 @@ public abstract class Compactor {
}
public CompletableFuture<Long> compact(String topic) {
- return RawReader.create(pulsar, topic, COMPACTION_SUBSCRIPTION, false).thenComposeAsync(
+ return RawReader.create(pulsar, topic, COMPACTION_SUBSCRIPTION, false, false).thenComposeAsync(
this::compactAndCloseReader, scheduler);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerCloseTest.java
index 3a446101990..66fff46b660 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerCloseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ConsumerCloseTest.java
@@ -18,14 +18,19 @@
*/
package org.apache.pulsar.client.impl;
+import static org.testng.Assert.assertTrue;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
@@ -76,7 +81,7 @@ public class ConsumerCloseTest extends ProducerConsumerBase {
.subscribe();
Assert.fail("Should have thrown an exception");
} catch (PulsarClientException e) {
- Assert.assertTrue(e.getCause() instanceof InterruptedException);
+ assertTrue(e.getCause() instanceof InterruptedException);
}
});
startConsumer.start();
@@ -88,4 +93,39 @@ public class ConsumerCloseTest extends ProducerConsumerBase {
Assert.assertEquals(clientImpl.consumersCount(), 0);
});
}
+
+ @Test
+ public void testReceiveWillDoneAfterClosedConsumer() throws Exception {
+ String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ String subName = "test-sub";
+ admin.topics().createNonPartitionedTopic(tpName);
+ admin.topics().createSubscription(tpName, subName, MessageId.earliest);
+ ConsumerImpl<byte[]> consumer =
+ (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(tpName).subscriptionName(subName).subscribe();
+ CompletableFuture<Message<byte[]>> future = consumer.receiveAsync();
+ consumer.close();
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(future.isDone());
+ });
+ }
+
+ @Test
+ public void testReceiveWillDoneAfterTopicDeleted() throws Exception {
+ String namespace = "public/default";
+ admin.namespaces().setAutoTopicCreation(namespace, AutoTopicCreationOverride.builder()
+ .allowAutoTopicCreation(false).build());
+ String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ String subName = "test-sub";
+ admin.topics().createNonPartitionedTopic(tpName);
+ admin.topics().createSubscription(tpName, subName, MessageId.earliest);
+ ConsumerImpl<byte[]> consumer =
+ (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(tpName).subscriptionName(subName).subscribe();
+ CompletableFuture<Message<byte[]>> future = consumer.receiveAsync();
+ admin.topics().delete(tpName, true);
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(future.isDone());
+ });
+ // cleanup.
+ admin.namespaces().removeAutoTopicCreation(namespace);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
index d9ddc00b2e8..a88d590f202 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java
@@ -27,12 +27,17 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -44,6 +49,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.api.RawReader;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
@@ -54,12 +60,17 @@ import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.protocol.Commands;
import org.awaitility.Awaitility;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import static org.apache.pulsar.client.impl.RawReaderImpl.DEFAULT_RECEIVER_QUEUE_SIZE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
@Test(groups = "broker-impl")
@Slf4j
@@ -215,7 +226,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
consumerConfiguration.setReadCompacted(true);
consumerConfiguration.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
consumerConfiguration.setAckReceiptEnabled(true);
- RawReader reader = RawReader.create(pulsarClient, consumerConfiguration, true).get();
+ RawReader reader = RawReader.create(pulsarClient, consumerConfiguration, true, true).get();
MessageId lastMessageId = reader.getLastMessageIdAsync().get();
while (true) {
@@ -547,11 +558,62 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
String topic2 = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader");
try {
- reader = RawReader.create(pulsarClient, topic2, subscription, false).get();
+ reader = RawReader.create(pulsarClient, topic2, subscription, false, true).get();
Assert.fail();
} catch (Exception e) {
Assert.assertTrue(e.getCause() instanceof PulsarClientException.TopicDoesNotExistException);
}
reader.closeAsync().join();
}
+
+ @Test(timeOut = 60000)
+ public void testReconnectsWhenServiceNotReady() throws Exception {
+ String topic = "persistent://my-property/my-ns/" + BrokerTestUtil.newUniqueName("reader");
+ String subscriptionName = "s1";
+ admin.topics().createNonPartitionedTopic(topic);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
+
+ // Inject a delay event for topic close, which leads to that the raw-reader will get a ServiceNotReady error,
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(topic, false).get().get();
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+ ManagedCursor compactionCursor = ml.openCursor(subscriptionName);
+ ManagedCursor spyCompactionCursor = spy(compactionCursor);
+ CountDownLatch delayCloseCursorSignal = new CountDownLatch(1);
+ Answer answer = new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+ delayCloseCursorSignal.await();
+ return invocationOnMock.callRealMethod();
+ }
+ };
+ doAnswer(answer).when(spyCompactionCursor).asyncClose(any(AsyncCallbacks.CloseCallback.class), any());
+ ml.getCursors().removeCursor(subscriptionName);
+ ml.getCursors().add(spyCompactionCursor, ml.getLastConfirmedEntry());
+
+ // Unload topic after reader is connected.
+ // The topic state comes to "fenced", then RawReader will get a ServiceNotReady error,
+ CompletableFuture<RawMessage> msgFuture = reader.readNextAsync();
+ CompletableFuture<Void> unloadFuture = admin.topics().unloadAsync(topic);
+ Awaitility.await().untilAsserted(() -> {
+ Assert.assertTrue(persistentTopic.isFenced());
+ });
+
+ // Verify: RasReader reconnected after that the unloading is finished, and it can consume successfully.
+ delayCloseCursorSignal.countDown();
+ unloadFuture.get();
+ MessageIdImpl msgIdSent = (MessageIdImpl) producer.send("msg");
+ RawMessage rawMessage = msgFuture.get();
+ Assert.assertNotNull(rawMessage);
+ MessageIdImpl msgIdReceived = (MessageIdImpl) rawMessage.getMessageId();
+ Assert.assertEquals(msgIdSent.getLedgerId(), msgIdReceived.getLedgerId());
+ Assert.assertEquals(msgIdSent.getEntryId(), msgIdReceived.getEntryId());
+
+ // cleanup.
+ rawMessage.close();;
+ producer.close();
+ reader.closeAsync().get();
+ admin.topics().delete(topic, false);
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
index ff9026dbba6..f43e2a1c672 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java
@@ -20,7 +20,11 @@ package org.apache.pulsar.compaction;
import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName;
import static org.apache.pulsar.broker.BrokerTestUtil.spyWithoutRecordingInvocations;
+import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -60,9 +64,11 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.api.OpenBuilder;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.commons.lang3.tuple.Pair;
@@ -101,6 +107,8 @@ import org.apache.pulsar.common.protocol.Markers;
import org.apache.pulsar.common.util.FutureUtil;
import org.awaitility.Awaitility;
import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
@@ -2320,6 +2328,65 @@ public class CompactionTest extends MockedPulsarServiceBaseTest {
producer.close();
}
+ @Test(timeOut = 120 * 1000)
+ public void testConcurrentCompactionAndTopicDelete() throws Exception {
+ final String topicName = newUniqueName("persistent://my-tenant/my-ns/tp");
+ admin.topics().createNonPartitionedTopic(topicName);
+ // Load up the topic.
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+
+ // Inject a reading delay to the compaction task,
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+ ManagedCursor compactionCursor = ml.openCursor(COMPACTION_SUBSCRIPTION);
+ ManagedCursor spyCompactionCursor = spy(compactionCursor);
+ CountDownLatch delayReadSignal = new CountDownLatch(1);
+ Answer answer = new Answer() {
+ @Override
+ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
+ delayReadSignal.await();
+ return invocationOnMock.callRealMethod();
+ }
+ };
+ doAnswer(answer).when(spyCompactionCursor).asyncReadEntries(anyInt(),
+ any(AsyncCallbacks.ReadEntriesCallback.class), any(), any(Position.class));
+ doAnswer(answer).when(spyCompactionCursor).asyncReadEntries(anyInt(), anyLong(),
+ any(AsyncCallbacks.ReadEntriesCallback.class), any(), any(Position.class));
+ doAnswer(answer).when(spyCompactionCursor).asyncReadEntriesOrWait(anyInt(), anyLong(),
+ any(AsyncCallbacks.ReadEntriesCallback.class), any(), any(Position.class));
+ ml.getCursors().removeCursor(COMPACTION_SUBSCRIPTION);
+ ml.getCursors().add(spyCompactionCursor, ml.getLastConfirmedEntry());
+
+ // Trigger a compaction task.
+ for (int i = 0; i < 2000; i++) {
+ producer.newMessage().key(String.valueOf(i)).value(String.valueOf(i)).send();
+ }
+ ConsumerImpl<String> consumer = (ConsumerImpl<String>) pulsarClient.newConsumer(Schema.STRING)
+ .topic(topicName).readCompacted(true).subscriptionName("s1")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ persistentTopic.triggerCompaction();
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(persistentTopic.getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 1);
+ });
+
+ // Since we injected a delay reading, the compaction task started and not finish yet.
+ // Call topic deletion, they two tasks are concurrently executed.
+ producer.close();
+ consumer.close();
+ CompletableFuture<Void> deleteTopicFuture = persistentTopic.deleteForcefully();
+
+ // Remove the injection after 3s.
+ Thread.sleep(3000);
+ delayReadSignal.countDown();
+
+ // Verify: topic deletion is successfully executed.
+ Awaitility.await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertTrue(deleteTopicFuture.isDone());
+ });
+ }
+
@Test
public void testEarliestSubsAfterRollover() throws Exception {
final String topicName = newUniqueName("persistent://my-tenant/my-ns/testEarliestSubsAfterRollover");
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
index b2c9b2b697b..bf32014af7a 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
@@ -362,6 +362,17 @@ public class PulsarClientException extends IOException {
}
}
+ /**
+ * Relates to server-side errors:
+ * ServiceUnitNotReadyException, TopicFencedException and SubscriptionFencedException.
+ */
+ public static class ServiceNotReadyException extends LookupException {
+
+ public ServiceNotReadyException(String msg) {
+ super(msg);
+ }
+ }
+
/**
* Connect exception thrown by Pulsar client.
*/
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index 24fd0034eb0..0563fa7e666 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -1353,7 +1353,7 @@ public class ClientCnx extends PulsarHandler {
case PersistenceError:
return new PulsarClientException.BrokerPersistenceException(errorMsg);
case ServiceNotReady:
- return new PulsarClientException.LookupException(errorMsg);
+ return new PulsarClientException.ServiceNotReadyException(errorMsg);
case TooManyRequests:
return new PulsarClientException.TooManyRequestsException(errorMsg);
case ProducerBlockedQuotaExceededError:
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
index fb5b6788d08..a4e262b35ae 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
@@ -64,7 +64,15 @@ public class ConnectionHandler {
* @apiNote If the returned future is completed exceptionally, reconnectLater will be called.
*/
CompletableFuture<Void> connectionOpened(ClientCnx cnx);
- default void connectionFailed(PulsarClientException e) {
+
+ /**
+ *
+ * @param e What error happened when tries to get a connection
+ * @return If "true", the connection handler will retry to get a connection, otherwise, it stops to get a new
+ * connection. If it returns "false", you should release resources that consumers/producers occupied.
+ */
+ default boolean connectionFailed(PulsarClientException e) {
+ return true;
}
}
@@ -142,22 +150,24 @@ public class ConnectionHandler {
}
private Void handleConnectionError(Throwable exception) {
+ boolean toRetry = true;
try {
log.warn("[{}] [{}] Error connecting to broker: {}",
state.topic, state.getHandlerName(), exception.getMessage());
if (exception instanceof PulsarClientException) {
- connection.connectionFailed((PulsarClientException) exception);
+ toRetry = connection.connectionFailed((PulsarClientException) exception);
} else if (exception.getCause() instanceof PulsarClientException) {
- connection.connectionFailed((PulsarClientException) exception.getCause());
+ toRetry = connection.connectionFailed((PulsarClientException) exception.getCause());
} else {
- connection.connectionFailed(new PulsarClientException(exception));
+ toRetry = connection.connectionFailed(new PulsarClientException(exception));
}
} catch (Throwable throwable) {
log.error("[{}] [{}] Unexpected exception after the connection",
state.topic, state.getHandlerName(), throwable);
}
-
- reconnectLater(exception);
+ if (toRetry) {
+ reconnectLater(exception);
+ }
return null;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 275966c26b3..f575a127c96 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -974,6 +974,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (e.getCause() instanceof PulsarClientException
&& PulsarClientException.isRetriableError(e.getCause())
+ && !isUnrecoverableError(e.getCause())
&& System.currentTimeMillis() < SUBSCRIBE_DEADLINE_UPDATER.get(ConsumerImpl.this)) {
future.completeExceptionally(e.getCause());
} else if (!subscribeFuture.isDone()) {
@@ -985,18 +986,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
+ "with subscription name %s when connecting to the broker",
topicName.toString(), subscription)));
client.cleanupConsumer(this);
- } else if (e.getCause() instanceof TopicDoesNotExistException) {
- // The topic was deleted after the consumer was created, and we're
- // not allowed to recreate the topic. This can happen in few cases:
- // * Regex consumer getting error after topic gets deleted
- // * Regular consumer after topic is manually delete and with
- // auto-topic-creation set to false
- // No more retries are needed in this case.
- setState(State.Failed);
- closeConsumerTasks();
- client.cleanupConsumer(this);
- log.warn("[{}][{}] Closed consumer because topic does not exist anymore {}",
- topic, subscription, cnx.channel().remoteAddress());
+ } else if (isUnrecoverableError(e.getCause())) {
+ closeWhenReceivedUnrecoverableError(e.getCause(), cnx);
} else {
// consumer was subscribed and connected but we got some error, keep trying
future.completeExceptionally(e.getCause());
@@ -1011,6 +1002,37 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return future;
}
+ /***
+ * Different consumer implementation can define its additional unrecoverable error.
+ */
+ protected boolean isUnrecoverableError(Throwable t) {
+ // TopicDoesNotExistException: topic has been deleted.
+ // NotFoundException: topic has been deleted.
+ // IllegalStateException: consumer has been closed.
+ return (t instanceof TopicDoesNotExistException) || (t instanceof IllegalStateException)
+ || (t instanceof PulsarClientException.NotFoundException);
+ }
+
+ protected void closeWhenReceivedUnrecoverableError(Throwable t, ClientCnx cnx) {
+ // The topic was deleted after the consumer was created, and we're
+ // not allowed to recreate the topic. This can happen in few cases:
+ // * Regex consumer getting error after topic gets deleted
+ // * Regular consumer after topic is manually delete and with
+ // auto-topic-creation set to false
+ // No more retries are needed in this case.
+ final String cnxStr = cnx == null ? "null" : String.valueOf(cnx.channel().remoteAddress());
+ log.warn("[{}][{}] {} Closed consumer because get an error that does not support to retry: {} {}",
+ topic, subscription, cnxStr, t.getClass().getName(), t.getMessage());
+ closeAsync().whenComplete((__, ex) -> {
+ if (ex == null) {
+ setState(State.Failed);
+ return;
+ }
+ log.error("[{}][{}] {} Failed to close consumer after got an error that does not support to retry: {} {}",
+ topic, subscription, cnxStr, t.getClass().getName(), t.getMessage());
+ });
+ }
+
protected void consumerIsReconnectedToBroker(ClientCnx cnx, int currentQueueSize) {
log.info("[{}][{}] Subscribed to topic on {} -- consumer: {}", topic, subscription,
cnx.channel().remoteAddress(), consumerId);
@@ -1091,7 +1113,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
@Override
- public void connectionFailed(PulsarClientException exception) {
+ public boolean connectionFailed(PulsarClientException exception) {
boolean nonRetriableError = !PulsarClientException.isRetriableError(exception);
boolean timeout = System.currentTimeMillis() > lookupDeadline;
if (nonRetriableError || timeout) {
@@ -1107,10 +1129,18 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
closeConsumerTasks();
deregisterFromClientCnx();
client.cleanupConsumer(this);
+ return false;
+ } else {
+ Throwable actError = FutureUtil.unwrapCompletionException(exception);
+ if (isUnrecoverableError(actError)) {
+ closeWhenReceivedUnrecoverableError(actError, null);
+ return false;
+ }
}
} else {
previousExceptionCount.incrementAndGet();
}
+ return true;
}
@Override
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 50a6f9aa81b..e9decbfa0f5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -1965,13 +1965,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
// Close the producer since topic does not exist.
- if (cause instanceof PulsarClientException.TopicDoesNotExistException) {
- closeAsync().whenComplete((v, ex) -> {
- if (ex != null) {
- log.error("Failed to close producer on TopicDoesNotExistException.", ex);
- }
- producerCreatedFuture.completeExceptionally(cause);
- });
+ if (isUnrecoverableError(cause)) {
+ closeWhenReceivedUnrecoverableError(cause, cnx);
future.complete(null);
return null;
}
@@ -2043,8 +2038,28 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
return future;
}
+ protected boolean isUnrecoverableError(Throwable t) {
+ // TopicDoesNotExistException: topic has been deleted.
+ // NotFoundException: topic has been deleted.
+ // IllegalStateException: producer has been closed.
+ return (t instanceof PulsarClientException.TopicDoesNotExistException) || (t instanceof IllegalStateException)
+ || (t instanceof PulsarClientException.NotFoundException);
+ }
+
+ protected void closeWhenReceivedUnrecoverableError(Throwable t, ClientCnx cnx) {
+ final String cnxStr = cnx == null ? "null" : String.valueOf(cnx.channel().remoteAddress());
+ log.warn("[{}][{}] {} Closed producer because get an error that does not support to retry: {} {}",
+ topic, producerName, cnxStr, t.getClass().getName(), t.getMessage());
+ closeAsync().whenComplete((v, ex) -> {
+ if (ex != null) {
+ log.error("Failed to close producer on TopicDoesNotExistException.", ex);
+ }
+ producerCreatedFuture.completeExceptionally(t);
+ });
+ }
+
@Override
- public void connectionFailed(PulsarClientException exception) {
+ public boolean connectionFailed(PulsarClientException exception) {
boolean nonRetriableError = !PulsarClientException.isRetriableError(exception);
boolean timeout = System.currentTimeMillis() > lookupDeadline;
if (nonRetriableError || timeout) {
@@ -2059,10 +2074,18 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
closeProducerTasks();
setState(State.Failed);
client.cleanupProducer(this);
+ return false;
+ } else {
+ Throwable actError = FutureUtil.unwrapCompletionException(exception);
+ if (isUnrecoverableError(actError)) {
+ closeWhenReceivedUnrecoverableError(actError, null);
+ return false;
+ }
}
} else {
previousExceptionCount.incrementAndGet();
}
+ return true;
}
private void closeProducerTasks() {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java
index 93fa7082f33..5357b154622 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicListWatcher.java
@@ -89,7 +89,7 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
}
@Override
- public void connectionFailed(PulsarClientException exception) {
+ public boolean connectionFailed(PulsarClientException exception) {
boolean nonRetriableError = !PulsarClientException.isRetriableError(exception);
if (nonRetriableError) {
exception.setPreviousExceptionCount(previousExceptionCount);
@@ -98,10 +98,12 @@ public class TopicListWatcher extends HandlerState implements ConnectionHandler.
log.info("[{}] Watcher creation failed for {} with non-retriable error {}",
topic, name, exception.getMessage());
deregisterFromClientCnx();
+ return false;
}
} else {
previousExceptionCount.incrementAndGet();
}
+ return true;
}
@Override
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java
index a1fe78d7290..4ca742d98ea 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TransactionMetaStoreHandler.java
@@ -122,7 +122,7 @@ public class TransactionMetaStoreHandler extends HandlerState
}
@Override
- public void connectionFailed(PulsarClientException exception) {
+ public boolean connectionFailed(PulsarClientException exception) {
boolean nonRetriableError = !PulsarClientException.isRetriableError(exception);
boolean timeout = System.currentTimeMillis() > lookupDeadline;
if (nonRetriableError || timeout) {
@@ -136,10 +136,12 @@ public class TransactionMetaStoreHandler extends HandlerState
+ "timeout", transactionCoordinatorId, exception);
}
setState(State.Failed);
+ return false;
}
} else {
previousExceptionCount.getAndIncrement();
}
+ return true;
}
@Override
(pulsar) 03/08: [improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (#24386)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7ed6edc8eddffb01525a0a52d16063dc47c8c4da
Author: Malla Sandeep <sa...@gmail.com>
AuthorDate: Fri Jun 6 01:57:49 2025 +0530
[improve][broker] Added synchronized for sendMessages in Non-Persistent message dispatchers (#24386)
(cherry picked from commit cb0fe5b0f65f09937e136eda9afeef13a7b08a03)
---
.../nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java | 2 +-
.../NonPersistentStickyKeyDispatcherMultipleConsumers.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
index ec83b4590b7..414e9235418 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
@@ -56,7 +56,7 @@ public final class NonPersistentDispatcherSingleActiveConsumer extends AbstractD
}
@Override
- public void sendMessages(List<Entry> entries) {
+ public synchronized void sendMessages(List<Entry> entries) {
Consumer currentConsumer = getActiveConsumer();
if (currentConsumer != null && currentConsumer.getAvailablePermits() > 0 && currentConsumer.isWritable()) {
SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
index ce674cf471e..915f9c0f925 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
@@ -135,7 +135,7 @@ public class NonPersistentStickyKeyDispatcherMultipleConsumers extends NonPersis
};
@Override
- public void sendMessages(List<Entry> entries) {
+ public synchronized void sendMessages(List<Entry> entries) {
if (entries.isEmpty()) {
return;
}
(pulsar) 08/08: [fix][ml]Revert a behavior change of releasing idle offloaded ledger handle: only release idle BlobStoreBackedReadHandle (#24384)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9adbbbf8f3907930eb12d4131f84bdae8d9ea6f9
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Mon Jun 9 21:33:46 2025 +0800
[fix][ml]Revert a behavior change of releasing idle offloaded ledger handle: only release idle BlobStoreBackedReadHandle (#24384)
(cherry picked from commit e0d7faa087f7ae6895f139b89ef333c8211f71d3)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 ++-
.../mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java | 2 +-
2 files changed, 3 insertions(+), 2 deletions(-)
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 ede21e0beea..7a6d7b884bd 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
@@ -2663,7 +2663,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
ReadHandle readHandle = ledger.join();
if (readHandle instanceof OffloadedLedgerHandle offloadedLedgerHandle) {
int pendingRead = offloadedLedgerHandle.getPendingRead();
- if (pendingRead == 0) {
+ long lastAccessTimestamp = offloadedLedgerHandle.lastAccessTimestamp();
+ if (lastAccessTimestamp >= 0 && pendingRead == 0) {
long delta = now - offloadedLedgerHandle.lastAccessTimestamp();
if (delta >= inactiveOffloadedLedgerEvictionTimeMs) {
log.info("[{}] Offloaded ledger {} can be released ({} ms elapsed since last access)",
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
index 84d9d118728..916e33c3642 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java
@@ -135,8 +135,8 @@ public class BlobStoreBackedReadHandleImpl implements ReadHandle, OffloadedLedge
// is better.
PENDING_READ_UPDATER.incrementAndGet(this);
promise.whenComplete((__, ex) -> {
- PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
lastAccessTimestamp = System.currentTimeMillis();
+ PENDING_READ_UPDATER.decrementAndGet(BlobStoreBackedReadHandleImpl.this);
});
executor.execute(() -> {
if (state == State.Closed) {