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 2026/07/25 14:17:56 UTC
(pulsar) branch branch-4.2 updated (7ecf63ec2e9f -> e5c0a82b8f47)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 7ecf63ec2e9f [improve][build] Upgrade docker base image Alpine to 3.24 (#26225)
new 07d214e41b17 [fix][ml] Preserve ledger properties when closing ledger (#26227)
new 242202b1e318 [fix][broker] Check deliverAt before containsMessage in bucket addMessage (#26230)
new d7a7a6881fb7 [fix][broker] Trigger max read position callback for messages published during transaction buffer recovery (#26234)
new 83af298f43f7 [improve][misc] Upgrade Jetty to 12.1.11 (#26233)
new 8cbbb6e6e59f [fix][broker] Prevent stale read completions from stranding Failover subscriptions (#26174)
new 1d9dadaba53b [fix][ml] Preserve ledger entries/size when transformLedgerInfo callback completes after a concurrent close (#26228)
new 36176936220b [improve][offload] Support credentials from offload policies for S3 and Aliyun OSS drivers (#26232)
new d195ab3f948e [fix][meta] Fix RocksdbMetadataStore instanceId not advancing across restarts (#26218)
new f066cfe8a09e [fix][broker] Fix Key_Shared delivery stall when look-ahead triggers at the end of the topic (#26236)
new b3682742836a [fix][meta] Record get op stats on the correct completion branch in AbstractMetadataStore (#26201)
new e5c0a82b8f47 [fix][meta] Complete handleMetadataEvent future exceptionally when the initial get fails (#26199)
The 11 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:
distribution/server/src/assemble/LICENSE.bin.txt | 72 +--
distribution/shell/src/assemble/LICENSE.bin.txt | 24 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 30 +-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 240 +++++++-
pom.xml | 2 +-
.../bucket/BucketDelayedDeliveryTracker.java | 8 +-
.../PersistentDispatcherSingleActiveConsumer.java | 39 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 13 +-
.../buffer/impl/TopicTransactionBuffer.java | 21 +-
.../bucket/BucketDelayedDeliveryTrackerTest.java | 48 ++
.../pulsar/broker/service/PersistentTopicTest.java | 2 +-
...ispatcherSingleActiveConsumerStuckReadTest.java | 673 +++++++++++++++++++++
...rsistentDispatcherSingleActiveConsumerTest.java | 2 +-
...ntStickyKeyDispatcherMultipleConsumersTest.java | 99 +++
.../impl/TopicTransactionBufferRecoveryTest.java | 144 +++++
.../common/policies/data/OffloadPoliciesImpl.java | 24 +-
.../common/policies/data/OffloadPoliciesTest.java | 46 ++
.../metadata/impl/AbstractMetadataStore.java | 12 +-
.../pulsar/metadata/impl/RocksdbMetadataStore.java | 2 +-
.../impl/AbstractMetadataStoreGetStatsTest.java | 122 ++++
.../impl/MetadataEventSynchronizerTest.java | 31 +
.../jcloud/provider/JCloudBlobStoreProvider.java | 19 +
.../provider/JCloudBlobStoreProviderTest.java | 46 ++
23 files changed, 1622 insertions(+), 97 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
create mode 100644 pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/AbstractMetadataStoreGetStatsTest.java
(pulsar) 11/11: [fix][meta] Complete handleMetadataEvent future exceptionally when the initial get fails (#26199)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e5c0a82b8f4791b74a71acf3705c9aadafe10b25
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Sat Jul 25 20:40:47 2026 +0800
[fix][meta] Complete handleMetadataEvent future exceptionally when the initial get fails (#26199)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit bc56ba8d30b2a96648a041359579b1e1138fb6f3)
---
.../metadata/impl/AbstractMetadataStore.java | 10 ++++---
.../impl/MetadataEventSynchronizerTest.java | 31 ++++++++++++++++++++++
2 files changed, 38 insertions(+), 3 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
index a463e0f1aec9..ffe576c6ecd6 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
@@ -212,14 +212,14 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
@Override
public CompletableFuture<Void> handleMetadataEvent(MetadataEvent event) {
CompletableFuture<Void> result = new CompletableFuture<>();
- get(event.getPath()).thenApply(res -> {
+ get(event.getPath()).thenAccept(res -> {
Set<CreateOption> options = event.getOptions() != null ? event.getOptions()
: Collections.emptySet();
if (res.isPresent()) {
GetResult existingValue = res.get();
if (shouldIgnoreEvent(event, existingValue)) {
result.complete(null);
- return result;
+ return;
}
}
// else update the event
@@ -241,7 +241,11 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
}
return false;
});
- return result;
+ }).exceptionally(ex -> {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.warn("Failed to handle metadata event {}", event.getPath(), cause);
+ result.completeExceptionally(cause);
+ return null;
});
return result;
}
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataEventSynchronizerTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataEventSynchronizerTest.java
index 3b07e0b3c2bd..90c17b7c5ae4 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataEventSynchronizerTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataEventSynchronizerTest.java
@@ -20,13 +20,22 @@ package org.apache.pulsar.metadata.impl;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
import java.util.Optional;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataEvent;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
+import org.apache.pulsar.metadata.api.NotificationType;
import org.awaitility.Awaitility;
import org.testng.annotations.Test;
@@ -75,6 +84,28 @@ public class MetadataEventSynchronizerTest {
});
}
+ @Test
+ public void testHandleMetadataEventCompletesWhenGetFails() throws Exception {
+ @Cleanup
+ LocalMemoryMetadataStore store = new LocalMemoryMetadataStore("memory:local",
+ MetadataStoreConfig.builder().build()) {
+ @Override
+ public CompletableFuture<Optional<GetResult>> storeGet(String path) {
+ return CompletableFuture.failedFuture(
+ new MetadataStoreException("injected storeGet failure"));
+ }
+ };
+
+ MetadataEvent event = new MetadataEvent("/test", "value".getBytes(StandardCharsets.UTF_8),
+ new HashSet<>(), null, System.currentTimeMillis(), "test-cluster", NotificationType.Modified);
+
+ CompletableFuture<Void> result = store.handleMetadataEvent(event);
+ // The future must not hang when the initial get() fails: it should complete exceptionally
+ ExecutionException ex = expectThrows(ExecutionException.class, () -> result.get(5, TimeUnit.SECONDS));
+ assertTrue(ex.getCause() instanceof MetadataStoreException,
+ "expected MetadataStoreException cause but got: " + ex.getCause());
+ }
+
@Test
public void testPathValid() {
assertFalse(AbstractMetadataStore.isValidPath(null));
(pulsar) 01/11: [fix][ml] Preserve ledger properties when closing ledger (#26227)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 07d214e41b1738545bc34c7f3d86ca4c8b541b7c
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Thu Jul 23 12:42:53 2026 +0800
[fix][ml] Preserve ledger properties when closing ledger (#26227)
(cherry picked from commit 79a2c6ff5641fb6fae01576c51d730ed59cdebe1)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 12 ++++++--
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 36 ++++++++++++----------
2 files changed, 28 insertions(+), 20 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 d5462bd76ebb..0d4530e11e86 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
@@ -1971,9 +1971,15 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.debug("[{}] Ledger has been closed id={} entries={}", name, lh.getId(), entriesInLedger);
}
if (entriesInLedger > 0) {
- LedgerInfo info = new LedgerInfo().setLedgerId(lh.getId()).setEntries(entriesInLedger)
- .setSize(lh.getLength()).setTimestamp(clock.millis());
- ledgers.put(lh.getId(), info);
+ ledgers.compute(lh.getId(), (ledgerId, oldInfo) -> {
+ LedgerInfo info = new LedgerInfo();
+ if (oldInfo != null) {
+ info.copyFrom(oldInfo);
+ } else {
+ info.setLedgerId(ledgerId);
+ }
+ return info.setEntries(entriesInLedger).setSize(lh.getLength()).setTimestamp(clock.millis());
+ });
} else {
// The last ledger was empty, so we can discard it
ledgers.remove(lh.getId());
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index ff75e77b9529..b8ae552ff6ab 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -4668,30 +4668,27 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
private ManagedLedger testSetLedgerProperty0(String name) throws Exception {
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setMaxEntriesPerLedger(2);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(name, config);
ml.addEntry("entry-1".getBytes());
+ long firstLedger = ml.ledgers.firstKey();
+ ml.asyncAddLedgerProperty(firstLedger, "key1", "value1").join();
ml.addEntry("entry-2".getBytes());
ml.addEntry("entry-3".getBytes());
+ long lastLedger = ml.ledgers.lastKey();
+ ml.asyncAddLedgerProperty(lastLedger, "key2", "value2").join();
- Assert.assertEquals(ml.getLedgersInfo().size(), 2);
- long firstLedger = ml.getLedgersInfo().firstKey();
- LedgerInfo firstLedgerInfo = ml.getLedgersInfo().get(firstLedger);
- Assert.assertEquals(firstLedgerInfo.getPropertiesCount(), 0);
- long lastLedger = ml.getLedgersInfo().lastKey();
- LedgerInfo lastLedgerInfo = ml.getLedgersInfo().get(lastLedger);
- Assert.assertEquals(lastLedgerInfo.getPropertiesCount(), 0);
-
- ml.asyncAddLedgerProperty(firstLedger, "key1", "value1").get();
- ml.asyncAddLedgerProperty(lastLedger, "key2", "value2").get();
-
- firstLedgerInfo = ml.getLedgersInfo().get(firstLedger);
- Assert.assertEquals(firstLedgerInfo.getPropertiesCount(), 1);
- Assert.assertEquals(ml.asyncGetLedgerProperty(firstLedger, "key1").get(), "value1");
+ Assert.assertNotEquals(firstLedger, lastLedger);
+ LedgerInfo firstLedgerInfo = ml.ledgers.get(firstLedger);
+ Assert.assertEquals(firstLedgerInfo.getEntries(), 2);
- lastLedgerInfo = ml.getLedgersInfo().get(lastLedger);
- Assert.assertEquals(lastLedgerInfo.getPropertiesCount(), 1);
- Assert.assertEquals(ml.asyncGetLedgerProperty(lastLedger, "key2").get(), "value2");
+ ml.asyncAddLedgerProperty(firstLedger, "key3", "value3").join();
+ ml.asyncAddLedgerProperty(lastLedger, "key4", "value4").join();
+ Assert.assertEquals(ml.asyncGetLedgerProperty(firstLedger, "key1").join(), "value1");
+ Assert.assertEquals(ml.asyncGetLedgerProperty(firstLedger, "key3").join(), "value3");
+ Assert.assertEquals(ml.asyncGetLedgerProperty(lastLedger, "key2").join(), "value2");
+ Assert.assertEquals(ml.asyncGetLedgerProperty(lastLedger, "key4").join(), "value4");
return ml;
}
@@ -4705,6 +4702,11 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
ml.asyncRemoveLedgerProperty(firstLedger, "key1").get();
ml.asyncRemoveLedgerProperty(lastLedger, "key2").get();
+ Assert.assertEquals(ml.getLedgersInfo().get(firstLedger).getPropertiesCount(), 1);
+ Assert.assertEquals(ml.getLedgersInfo().get(lastLedger).getPropertiesCount(), 1);
+
+ ml.asyncRemoveLedgerProperty(firstLedger, "key3").get();
+ ml.asyncRemoveLedgerProperty(lastLedger, "key4").get();
Assert.assertEquals(ml.getLedgersInfo().get(firstLedger).getPropertiesCount(), 0);
Assert.assertEquals(ml.getLedgersInfo().get(lastLedger).getPropertiesCount(), 0);
}
(pulsar) 07/11: [improve][offload] Support credentials from offload policies for S3 and Aliyun OSS drivers (#26232)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 36176936220b4b019387a437fb622105071c090a
Author: kannar <ka...@gmail.com>
AuthorDate: Sat Jul 25 10:37:14 2026 +0200
[improve][offload] Support credentials from offload policies for S3 and Aliyun OSS drivers (#26232)
Co-authored-by: Claude Fable 5 <no...@anthropic.com>
(cherry picked from commit a965f5f7df3e060d9a9d4e63caf0561a2b90441c)
---
.../common/policies/data/OffloadPoliciesImpl.java | 24 +++++++----
.../common/policies/data/OffloadPoliciesTest.java | 46 ++++++++++++++++++++++
.../jcloud/provider/JCloudBlobStoreProvider.java | 19 +++++++++
.../provider/JCloudBlobStoreProviderTest.java | 46 ++++++++++++++++++++++
4 files changed, 127 insertions(+), 8 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java
index c5579d1008c7..911438ac0f83 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java
@@ -63,8 +63,14 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
CONFIGURATION_FIELDS = Collections.unmodifiableList(temp);
}
- public static final List<String> INTERNAL_SUPPORTED_DRIVER = List.of("S3",
- "aws-s3", "google-cloud-storage", "filesystem", "azureblob", "aliyun-oss");
+ public static final String DRIVER_S3 = "S3";
+ public static final String DRIVER_AWS_S3 = "aws-s3";
+ public static final String DRIVER_GOOGLE_CLOUD_STORAGE = "google-cloud-storage";
+ public static final String DRIVER_FILESYSTEM = "filesystem";
+ public static final String DRIVER_AZUREBLOB = "azureblob";
+ public static final String DRIVER_ALIYUN_OSS = "aliyun-oss";
+ public static final List<String> INTERNAL_SUPPORTED_DRIVER = List.of(DRIVER_S3,
+ DRIVER_AWS_S3, DRIVER_GOOGLE_CLOUD_STORAGE, DRIVER_FILESYSTEM, DRIVER_AZUREBLOB, DRIVER_ALIYUN_OSS);
public static final List<String> DRIVER_NAMES;
static {
String extraDrivers = System.getProperty("pulsar.extra.offload.drivers", "");
@@ -221,7 +227,8 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
.managedLedgerOffloadReadBufferSizeInBytes(readBufferSizeInBytes)
.managedLedgerOffloadedReadPriority(readPriority);
- if (driver.equalsIgnoreCase(DRIVER_NAMES.get(0)) || driver.equalsIgnoreCase(DRIVER_NAMES.get(1))) {
+ if (driver.equalsIgnoreCase(DRIVER_S3) || driver.equalsIgnoreCase(DRIVER_AWS_S3)
+ || driver.equalsIgnoreCase(DRIVER_ALIYUN_OSS)) {
if (role != null) {
builder.s3ManagedLedgerOffloadRole(role);
}
@@ -240,7 +247,7 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
.s3ManagedLedgerOffloadServiceEndpoint(endpoint)
.s3ManagedLedgerOffloadMaxBlockSizeInBytes(maxBlockSizeInBytes)
.s3ManagedLedgerOffloadReadBufferSizeInBytes(readBufferSizeInBytes);
- } else if (driver.equalsIgnoreCase(DRIVER_NAMES.get(2))) {
+ } else if (driver.equalsIgnoreCase(DRIVER_GOOGLE_CLOUD_STORAGE)) {
builder.gcsManagedLedgerOffloadRegion(region)
.gcsManagedLedgerOffloadBucket(bucket)
.gcsManagedLedgerOffloadMaxBlockSizeInBytes(maxBlockSizeInBytes)
@@ -310,22 +317,23 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies {
if (managedLedgerOffloadDriver == null) {
return false;
}
- return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(0))
- || managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(1));
+ return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_S3)
+ || managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_AWS_S3)
+ || managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_ALIYUN_OSS);
}
public boolean isGcsDriver() {
if (managedLedgerOffloadDriver == null) {
return false;
}
- return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(2));
+ return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_GOOGLE_CLOUD_STORAGE);
}
public boolean isFileSystemDriver() {
if (managedLedgerOffloadDriver == null) {
return false;
}
- return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(3));
+ return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_FILESYSTEM);
}
public boolean bucketValid() {
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java
index b3b02e46a58b..d0f00fb33573 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java
@@ -96,6 +96,52 @@ public class OffloadPoliciesTest {
Assert.assertEquals(offloadPolicies.getManagedLedgerOffloadThresholdInSeconds(), offloadThresholdInSeconds);
}
+ @Test
+ public void testAliyunOssConfiguration() {
+ final String driver = "aliyun-oss";
+ final String region = "test-region";
+ final String bucket = "test-bucket";
+ final String role = "test-role";
+ final String roleSessionName = "test-role-session-name";
+ final String credentialId = "test-credential-id";
+ final String credentialSecret = "test-credential-secret";
+ final String endPoint = "test-endpoint";
+ final Integer maxBlockSizeInBytes = 5 * M;
+ final Integer readBufferSizeInBytes = 2 * M;
+ final Long offloadThresholdInBytes = 10L * M;
+ final Long offloadThresholdInSeconds = 1000L;
+ final Long offloadDeletionLagInMillis = 5L * MIN;
+
+ OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.create(
+ driver,
+ region,
+ bucket,
+ endPoint,
+ role,
+ roleSessionName,
+ credentialId,
+ credentialSecret,
+ maxBlockSizeInBytes,
+ readBufferSizeInBytes,
+ offloadThresholdInBytes,
+ offloadThresholdInSeconds,
+ offloadDeletionLagInMillis,
+ OffloadedReadPriority.TIERED_STORAGE_FIRST
+ );
+
+ Assert.assertTrue(offloadPolicies.isS3Driver());
+ Assert.assertEquals(offloadPolicies.getManagedLedgerOffloadDriver(), driver);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadRegion(), region);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadBucket(), bucket);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadServiceEndpoint(), endPoint);
+ // The CLI create() path must carry the credentials under the s3-prefixed keys the
+ // S3-compatible aliyun-oss offloader reads, not silently drop them.
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadCredentialId(), credentialId);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadCredentialSecret(), credentialSecret);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadMaxBlockSizeInBytes(), maxBlockSizeInBytes);
+ Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadReadBufferSizeInBytes(), readBufferSizeInBytes);
+ }
+
@Test
public void testGcsConfiguration() {
final String driver = "google-cloud-storage";
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java
index f859f2b73284..b8d30935b8d8 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java
@@ -429,6 +429,25 @@ public enum JCloudBlobStoreProvider implements Serializable, ConfigValidation, B
};
static final CredentialBuilder S3_CREDENTIAL_BUILDER = (TieredStorageConfiguration config) -> {
+ if (config.getCredentials() != null) {
+ return;
+ }
+ // Credentials provided in the tiered storage configuration take
+ // precedence over the environment variables. Offload policies carry
+ // credentials under the s3-prefixed keys for every S3-compatible
+ // driver (see OffloadPoliciesImpl), so those are the keys accepted.
+ String configId = config.getConfigProperty(S3_ID_FIELD);
+ String configSecret = config.getConfigProperty(S3_SECRET_FIELD);
+ if (StringUtils.isNotBlank(configId) || StringUtils.isNotBlank(configSecret)) {
+ if (StringUtils.isBlank(configId) || StringUtils.isBlank(configSecret)) {
+ throw new IllegalArgumentException(
+ "Both " + S3_ID_FIELD + " and " + S3_SECRET_FIELD
+ + " must be set when providing offload credentials in the configuration");
+ }
+ Credentials credentials = new Credentials(configId, configSecret);
+ config.setProviderCredentials(() -> credentials);
+ return;
+ }
String accountName = System.getenv().getOrDefault("ACCESS_KEY_ID", "");
// For forward compatibility
if (StringUtils.isEmpty(accountName.trim())) {
diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java
index f45cb30ea4c2..7ad2eef99474 100644
--- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java
+++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java
@@ -19,8 +19,10 @@
package org.apache.bookkeeper.mledger.offload.jcloud.provider;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
import java.util.HashMap;
import java.util.Map;
+import org.jclouds.domain.Credentials;
import org.testng.annotations.Test;
public class JCloudBlobStoreProviderTest {
@@ -130,4 +132,48 @@ public class JCloudBlobStoreProviderTest {
TieredStorageConfiguration configuration = new TieredStorageConfiguration(map);
configuration.getProvider().validate(configuration);
}
+
+ // Offload policies (OffloadPoliciesImpl) carry credentials under the
+ // s3-prefixed keys for every S3-compatible driver
+ private TieredStorageConfiguration credentialConfig(String driver, String id, String secret) {
+ Map<String, String> map = new HashMap<>();
+ map.put("managedLedgerOffloadDriver", driver);
+ map.put("managedLedgerOffloadServiceEndpoint", "http://storage.service");
+ map.put("managedLedgerOffloadBucket", "test-bucket");
+ if (id != null) {
+ map.put("s3ManagedLedgerOffloadCredentialId", id);
+ }
+ if (secret != null) {
+ map.put("s3ManagedLedgerOffloadCredentialSecret", secret);
+ }
+ return new TieredStorageConfiguration(map);
+ }
+
+ private void assertCredentialsFromConfig(String driver) {
+ TieredStorageConfiguration configuration =
+ credentialConfig(driver, "config-access-id", "config-access-secret");
+ configuration.getProvider().buildCredentials(configuration);
+ assertNotNull(configuration.getProviderCredentials());
+ Credentials credentials = configuration.getProviderCredentials().get();
+ assertEquals(credentials.identity, "config-access-id");
+ assertEquals(credentials.credential, "config-access-secret");
+ }
+
+ @Test
+ public void s3CredentialsFromConfigTest() {
+ assertCredentialsFromConfig("S3");
+ }
+
+ @Test
+ public void aliyunOssCredentialsFromConfigTest() {
+ assertCredentialsFromConfig("aliyun-oss");
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class,
+ expectedExceptionsMessageRegExp = "Both s3ManagedLedgerOffloadCredentialId and "
+ + "s3ManagedLedgerOffloadCredentialSecret must be set.*")
+ public void s3PartialCredentialsFromConfigTest() {
+ TieredStorageConfiguration configuration = credentialConfig("S3", "config-access-id", null);
+ configuration.getProvider().buildCredentials(configuration);
+ }
}
(pulsar) 05/11: [fix][broker] Prevent stale read completions from stranding Failover subscriptions (#26174)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8cbbb6e6e59f5c37756781024e0db051b8889104
Author: Alexandre Burgoni <23...@users.noreply.github.com>
AuthorDate: Fri Jul 24 00:02:34 2026 +0200
[fix][broker] Prevent stale read completions from stranding Failover subscriptions (#26174)
Signed-off-by: Alexandre Burgoni <al...@clever.cloud>
(cherry picked from commit 0679b0d897de064bd6edfd47694f35a9ff44e80f)
---
.../PersistentDispatcherSingleActiveConsumer.java | 39 +-
.../pulsar/broker/service/PersistentTopicTest.java | 2 +-
...ispatcherSingleActiveConsumerStuckReadTest.java | 673 +++++++++++++++++++++
...rsistentDispatcherSingleActiveConsumerTest.java | 2 +-
4 files changed, 710 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 2420b2107ea0..5cc36df3f4b8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -72,6 +72,11 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
private Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
protected volatile boolean havePendingRead = false;
+ // Monotonic identity of the read that havePendingRead currently refers to. Bumped under this monitor
+ // whenever readMoreEntries issues a read; a completion mutates dispatcher state only when its captured
+ // epoch still matches, so a completion whose read was disowned by a redeliver-driven rewind + re-arm
+ // is ignored instead of clearing havePendingRead for the newer outstanding read. Guarded by "this".
+ private long readOpEpoch = 0L;
protected volatile int readBatchSize;
protected final Backoff readFailureBackoff;
@@ -155,7 +160,21 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
}
}
- private synchronized void readEntriesComplete(List<Entry> entries, Consumer readConsumer, long epoch) {
+ private synchronized void readEntriesComplete(List<Entry> entries, Consumer readConsumer, long epoch,
+ long readOpEpoch) {
+ if (readOpEpoch != this.readOpEpoch) {
+ // Stale completion: this read was disowned by a redeliver-driven cursor rewind + re-arm
+ // (internalRedeliverUnacknowledgedMessages) that already issued a newer read. Clearing
+ // havePendingRead or dispatching here would strand the newer armed read and double-deliver;
+ // release the entries (the newer read re-delivers from the rewound position) and do nothing else.
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Discarding stale read completion: size={}, staleReadOpEpoch={},"
+ + " currentReadOpEpoch={}",
+ name, readConsumer, entries.size(), readOpEpoch, this.readOpEpoch);
+ }
+ entries.forEach(Entry::release);
+ return;
+ }
if (log.isDebugEnabled()) {
log.debug("[{}-{}] Got messages: {}", name, readConsumer, entries.size());
}
@@ -368,6 +387,9 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
log.debug("[{}-{}] Schedule read of {} messages", name, consumer, messagesToRead);
}
havePendingRead = true;
+ // Tag this read so a completion that has since been superseded by a redeliver-driven re-arm
+ // can be detected as stale (see readEntriesComplete / readEntriesFailed).
+ final long readOpEpoch = ++this.readOpEpoch;
// TODO: should we pass the consumer epoch for compacted read path? See
// https://github.com/apache/pulsar/issues/13690
final var epoch = consumer.readCompacted() ? DEFAULT_CONSUMER_EPOCH : consumer.getConsumerEpoch();
@@ -384,9 +406,9 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
}
entriesFuture.whenCompleteAsync((entries, e) -> {
if (e == null) {
- readEntriesComplete(entries, consumer, epoch);
+ readEntriesComplete(entries, consumer, epoch, readOpEpoch);
} else {
- readEntriesFailed(e, consumer);
+ readEntriesFailed(e, consumer, readOpEpoch);
}
}, executor);
}
@@ -459,7 +481,16 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
}
@VisibleForTesting
- public synchronized void readEntriesFailed(Throwable throwable, Consumer consumer) {
+ public synchronized void readEntriesFailed(Throwable throwable, Consumer consumer, long readOpEpoch) {
+ if (readOpEpoch != this.readOpEpoch) {
+ // Stale failure for a read already disowned by a redeliver-driven re-arm; clearing havePendingRead
+ // or rescheduling here would disturb the newer outstanding read. Ignore it.
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Ignoring stale read failure: staleReadOpEpoch={}, currentReadOpEpoch={}",
+ name, consumer, readOpEpoch, this.readOpEpoch);
+ }
+ return;
+ }
havePendingRead = false;
final var exception = FutureUtil.unwrapCompletionException(throwable);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index 62c30dd65356..055bc3a5d1cf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -382,7 +382,7 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
PersistentDispatcherSingleActiveConsumer dispatcher = new PersistentDispatcherSingleActiveConsumer(cursor,
SubType.Exclusive, 1, topic, null);
Consumer consumer = mock(Consumer.class);
- dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), consumer);
+ dispatcher.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), consumer, 0L);
verify(topic, atLeast(1)).getBrokerService();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java
new file mode 100644
index 000000000000..1e893b9e597a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerStuckReadTest.java
@@ -0,0 +1,673 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Dispatcher;
+import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
+import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.testcontext.PulsarTestContext;
+import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
+import org.apache.pulsar.common.policies.data.stats.ConsumerStatsImpl;
+import org.awaitility.Awaitility;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Regression test for a single-active-consumer (Failover) subscription that becomes permanently stuck
+ * when a client redeliver command races the completion of an already-dispatched managed-ledger read.
+ *
+ * <p><b>Four coupled state variables.</b> A caught-up Failover subscription is described by four pieces
+ * of state that must stay consistent:
+ * <ol>
+ * <li>{@code cursor.waitingReadOp} — the armed tail-wait read op
+ * ({@link ManagedCursorImpl#hasPendingReadRequest()});</li>
+ * <li>the cursor's membership in {@link ManagedLedgerImpl}'s {@code waitingCursors} queue — the
+ * set a publish walks to wake parked readers;</li>
+ * <li>{@code cursor.pendingReadOps} — reads currently in flight
+ * ({@link ManagedCursorImpl#getPendingReadOpsCount()});</li>
+ * <li>{@code dispatcher.havePendingRead} — the dispatcher's belief that a read is outstanding.</li>
+ * </ol>
+ * In healthy operation an armed wait op implies {@code havePendingRead}, and implies the cursor is
+ * either queued in {@code waitingCursors} or has an in-flight read.
+ *
+ * <p><b>The race.</b>
+ * {@code PersistentDispatcherSingleActiveConsumer#internalRedeliverUnacknowledgedMessages} unconditionally
+ * clears {@code havePendingRead} and arms a fresh read via {@code readMoreEntries} <i>without</i> draining
+ * a read that {@code notifyEntriesAvailable} has already dispatched. The stale read's completion
+ * ({@code readEntriesComplete} / {@code readEntriesFailed}) then clears {@code havePendingRead} again:
+ * there is no read-generation guard on {@code havePendingRead}, and the only staleness check is the
+ * {@code readConsumer != currentConsumer} identity comparison in {@code readEntriesComplete}, which runs
+ * <i>after</i> that same method has already cleared the flag at its top. When the redeliver's re-arm and
+ * the stale completion interleave in the order [redeliver, then stale completion], the result is the seed
+ * state — an armed wait op with {@code havePendingRead == false}. This is still benign while the
+ * cursor remains in {@code waitingCursors}.
+ *
+ * <p><b>The absorbing conversion.</b> When the last consumer disconnects, the durable-cursor path
+ * ({@code AbstractDispatcherSingleActiveConsumer#removeConsumer} → {@code cancelPendingRead}, then
+ * {@code ManagedLedger#removeWaitingCursor}) deliberately does not cancel the pending read for a durable
+ * cursor — see the durable-cursor comment in {@code PersistentSubscription#removeConsumer}. Because
+ * {@code havePendingRead} is already {@code false}, {@code cancelPendingRead} short-circuits on its
+ * {@code havePendingRead} guard and leaves the wait op armed, while {@code removeWaitingCursor} strips the
+ * cursor from the queue. The subscription is now in the absorbing orphan state: an armed wait op, absent
+ * from {@code waitingCursors}, with no in-flight read and {@code havePendingRead == false}. Every
+ * subsequent re-arm CAS-fails on the {@code WAITING_READ_OP_UPDATER} compare-and-set in
+ * {@code ManagedCursorImpl#asyncReadEntriesWithSkipOrWait}, raising {@code ConcurrentWaitCallbackException}
+ * — which {@code readEntriesFailed} returns early on instead of rescheduling — and every
+ * publish's {@code notifyCursors} poll-misses. The subscription never reads again — which is exactly
+ * what both tests assert behaviourally: after the disconnect a fresh consumer reconnects, and a message
+ * published afterwards must actually reach that consumer's {@code sendMessages}.
+ *
+ * <p>The two racing events genuinely arrive on different threads: the read completion is posted to the
+ * dispatcher's ordered executor by the managed-ledger/BookKeeper completion chain (the
+ * {@code entriesFuture.whenCompleteAsync(..., executor)} hand-off at the end of {@code readMoreEntries}),
+ * while the redeliver is posted by {@code redeliverUnacknowledgedMessages} from the client-command thread.
+ * Their arrival order is a real race.
+ *
+ * <p><b>Fidelity notes / test seams.</b>
+ * <ul>
+ * <li>The cursor, ledger, subscription, dispatcher, {@code waitingReadOp} compare-and-set,
+ * {@code checkForNewEntries}, {@code notifyEntriesAvailable}, {@code notifyCursors},
+ * {@code cancelPendingReadRequest}, {@code addWaitingCursor}/{@code removeWaitingCursor} and
+ * {@code havePendingRead} management are all real production code.</li>
+ * <li>The consumer lifecycle runs through the real {@code PersistentSubscription#addConsumer} and
+ * {@code PersistentSubscription#removeConsumer(Consumer, boolean)}, so dispatcher removal,
+ * {@code deactivateCursor()} and {@code removeWaitingCursor()} happen in the production order. The
+ * spied dispatcher is handed to the subscription through the existing production test seam
+ * {@code PersistentSubscription#reuseOrCreateDispatcher}.</li>
+ * <li>Only the metadata-parsing filter step ({@code filterEntriesForConsumer}) is stubbed on a Mockito
+ * spy, because the entries this test publishes are raw bytes rather than serialized Pulsar messages.
+ * The delivery tail itself is real: {@code dispatchEntriesToConsumer} runs and calls
+ * {@code Consumer#sendMessages}, which the mock consumer records before releasing the entries and
+ * completing immediately — so the production {@code readMoreEntries} continuation is posted by
+ * the real listener. Neither seam touches the four state variables above.</li>
+ * <li>The dispatcher's ordered executor is replaced with a {@link ManualExecutor} (injected through a
+ * spied {@code getTopicOrderedExecutor()} scoped to dispatcher construction) so every dispatcher
+ * async hop becomes an explicit task the test drains in a chosen order.</li>
+ * <li>The consumer's ack of {@code m1} is telescoped: it is applied before the dispatcher delivery of
+ * {@code m1}'s in-flight read completes. This is what keeps the schedule deterministic — after
+ * the ack the redeliver's {@code rewind} lands at the tail with no backlog, so the re-arm is a pure
+ * in-memory tail-wait and no BookKeeper read escapes the {@link ManualExecutor} to race the drain.
+ * Without the ack the {@code rewind} re-exposes the unacked entry and {@code readMoreEntries} takes
+ * the immediate-read branch of {@code ManagedCursorImpl#asyncReadEntriesWithSkipOrWait} (the
+ * {@code hasMoreEntries()} fast path that delegates straight to {@code asyncReadEntriesWithSkip}),
+ * issuing a real async read whose completion is off-board. The staged state (mark-delete at the
+ * tail plus an already-dispatched read whose completion is still pending) is production-reachable
+ * via a redeliver→ack→redeliver sequence.</li>
+ * <li>{@code newEntriesCheckDelayInMillis} is pinned to {@code 0} so {@code checkForNewEntries} runs
+ * inline at arm time rather than on the ledger scheduler. This is a determinism pin only; the bug
+ * itself is not an artifact of the zero delay. With the telescoped ack in place, every
+ * dispatcher-visible async hop is then an explicit task on the {@link ManualExecutor} board that the
+ * test drains in a chosen order.</li>
+ * </ul>
+ *
+ * <p>Reproduces apache/pulsar#26164.
+ */
+@Slf4j
+public class PersistentDispatcherSingleActiveConsumerStuckReadTest extends MockedBookKeeperTestCase {
+
+ private static final String TOPIC = "persistent://prop/ns/sac-stuck-read";
+ private static final int BOARD_WAIT_SECONDS = 5;
+ private static final long QUIESCE_TIMEOUT_SECONDS = 10;
+ private static final int QUIESCE_IDLE_ROUNDS = 4;
+ private static final long DELIVERY_TIMEOUT_SECONDS = 5;
+
+ private PulsarTestContext pulsarTestContext;
+ private BrokerService brokerService;
+
+ // Real managed-ledger objects under test.
+ private ManagedLedgerImpl ledger;
+ private ManagedCursorImpl cursor;
+
+ // Real topic + subscription; the dispatcher is a Mockito spy so we can stub the entry filter step.
+ private PersistentTopic topic;
+ private PersistentSubscription subscription;
+ private PersistentDispatcherSingleActiveConsumer dispatcher;
+ private Consumer consumer;
+
+ // The dispatcher's ordered executor: a manual task board we drain in a chosen order.
+ private ManualExecutor dispatcherExecutor;
+
+ // Everything the dispatcher handed to a Consumer#sendMessages tail, in delivery order.
+ private final List<Delivery> deliveries = Collections.synchronizedList(new ArrayList<>());
+
+ private final AtomicLong msgCounter = new AtomicLong();
+
+ @Override
+ protected ManagedLedgerConfig initManagedLedgerConfig(ManagedLedgerConfig config) {
+ super.initManagedLedgerConfig(config);
+ // Inline checkForNewEntries: the +10ms task runs synchronously at arm time -> deterministic.
+ config.setNewEntriesCheckDelayInMillis(0);
+ config.setMaxEntriesPerLedger(1_000_000);
+ config.setRetentionTime(1, TimeUnit.HOURS);
+ config.setRetentionSizeInMB(-1);
+ return config;
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // Fixture
+ // -----------------------------------------------------------------------------------------------
+
+ /** Build a fresh topic/subscription/dispatcher on top of a real ledger + cursor. */
+ private void buildFixture(String ledgerName) throws Exception {
+ ServiceConfiguration svcConfig = new ServiceConfiguration();
+ svcConfig.setBrokerShutdownTimeoutMs(0L);
+ svcConfig.setLoadBalancerOverrideBrokerNicSpeedGbps(Optional.of(1.0d));
+ svcConfig.setClusterName("test");
+ svcConfig.setActiveConsumerFailoverDelayTimeMillis(0);
+ svcConfig.setSystemTopicEnabled(false);
+ svcConfig.setTopicLevelPoliciesEnabled(false);
+
+ // Real BrokerService via PulsarTestContext, but backed by the real MockedBookKeeper-based
+ // ManagedLedgerFactory so we open a REAL ledger + cursor.
+ pulsarTestContext = PulsarTestContext.builderForNonStartableContext()
+ .config(svcConfig)
+ .spyByDefault()
+ .managedLedgerClients(bkc, factory)
+ .build();
+ brokerService = pulsarTestContext.getBrokerService();
+
+ ledger = (ManagedLedgerImpl) factory.open(ledgerName, initManagedLedgerConfig(new ManagedLedgerConfig()));
+ topic = new PersistentTopic(TOPIC, ledger, brokerService);
+ // Open the cursor AFTER topic construction so the topic does not auto-create a second
+ // subscription sharing this cursor.
+ cursor = (ManagedCursorImpl) ledger.openCursor("sub");
+ // A real subscription that adopts the spied dispatcher through the production test seam, so
+ // addConsumer/removeConsumer drive the real consumer lifecycle. The returned reference must be
+ // qualified: the inherited PersistentSubscription.dispatcher field would otherwise shadow ours.
+ subscription = new PersistentSubscription(topic, "sub", cursor, false) {
+ @Override
+ protected Dispatcher reuseOrCreateDispatcher(Dispatcher existingDispatcher, Consumer newConsumer) {
+ return PersistentDispatcherSingleActiveConsumerStuckReadTest.this.dispatcher;
+ }
+ };
+
+ // Inject the manual dispatcher executor via getTopicOrderedExecutor().chooseThread(), scoped to
+ // just the dispatcher construction so unrelated broker work keeps using the real executor.
+ OrderedExecutor realTopicOrdered = brokerService.getTopicOrderedExecutor();
+ dispatcherExecutor = new ManualExecutor();
+ OrderedExecutor topicOrdered = mock(OrderedExecutor.class);
+ doReturn(dispatcherExecutor).when(topicOrdered).chooseThread();
+ doReturn(dispatcherExecutor).when(topicOrdered).chooseThread(any());
+ doReturn(topicOrdered).when(brokerService).getTopicOrderedExecutor();
+ dispatcher = spy(new PersistentDispatcherSingleActiveConsumer(cursor, SubType.Failover, -1, topic,
+ subscription));
+ doReturn(realTopicOrdered).when(brokerService).getTopicOrderedExecutor();
+
+ // The published payloads are raw bytes, not serialized Pulsar messages, so the metadata-parsing
+ // filter step is stubbed out. The rest of the delivery tail (dispatchEntriesToConsumer ->
+ // Consumer#sendMessages) is real production code.
+ doReturn(0).when(dispatcher).filterEntriesForConsumer(any(), any(), any(), any(), any(),
+ anyBoolean(), any());
+
+ deliveries.clear();
+ consumer = newMockConsumer("c1");
+ }
+
+ /**
+ * A mock consumer whose {@code sendMessages} tail records what the dispatcher delivered, releases the
+ * entries and completes immediately. The succeeded future notifies its listener inline, so the
+ * production {@code readMoreEntries} continuation lands on the manual board exactly as it does in
+ * production.
+ */
+ private Consumer newMockConsumer(String consumerName) {
+ Consumer mockConsumer = mock(Consumer.class);
+ doReturn(1000).when(mockConsumer).getAvailablePermits();
+ doReturn(1).when(mockConsumer).getAvgMessagesPerEntry();
+ doReturn(true).when(mockConsumer).isWritable();
+ doReturn(false).when(mockConsumer).readCompacted();
+ doReturn(false).when(mockConsumer).isPreciseDispatcherFlowControl();
+ doReturn(false).when(mockConsumer).isBlocked();
+ doReturn(consumerName).when(mockConsumer).consumerName();
+ doReturn(0).when(mockConsumer).getPriorityLevel();
+ doReturn(0L).when(mockConsumer).getConsumerEpoch();
+ doReturn(SubType.Failover).when(mockConsumer).subType();
+ // PersistentSubscription.removeConsumer folds the removed consumer's counters into the subscription.
+ doReturn(new ConsumerStatsImpl()).when(mockConsumer).getStats();
+ doAnswer(inv -> {
+ List<Entry> entries = inv.getArgument(0);
+ for (Entry entry : entries) {
+ deliveries.add(Delivery.of(consumerName, entry.getPosition()));
+ entry.release();
+ }
+ EntryBatchSizes batchSizes = inv.getArgument(1);
+ if (batchSizes != null) {
+ batchSizes.recyle();
+ }
+ EntryBatchIndexesAcks batchIndexesAcks = inv.getArgument(2);
+ if (batchIndexesAcks != null) {
+ batchIndexesAcks.recycle();
+ }
+ return ImmediateEventExecutor.INSTANCE.newSucceededFuture(null);
+ }).when(mockConsumer).sendMessages(any(), any(), any(), anyInt(), anyLong(), anyLong(), any(), anyLong());
+ return mockConsumer;
+ }
+
+ private void tearDownFixture() {
+ try {
+ if (cursor != null && !cursor.isClosed()) {
+ cursor.close();
+ }
+ } catch (Exception ignore) {
+ // best-effort cleanup
+ }
+ try {
+ if (ledger != null) {
+ ledger.close();
+ }
+ } catch (Exception ignore) {
+ // best-effort cleanup
+ }
+ try {
+ if (pulsarTestContext != null) {
+ pulsarTestContext.close();
+ }
+ } catch (Exception ignore) {
+ // best-effort cleanup
+ }
+ cursor = null;
+ ledger = null;
+ brokerService = null;
+ pulsarTestContext = null;
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void afterMethod() {
+ tearDownFixture();
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // Manual executor task board
+ // -----------------------------------------------------------------------------------------------
+
+ /** A deterministic executor: submitted Runnables queue up and run only when the test drains them. */
+ static final class ManualExecutor extends AbstractExecutorService {
+ private final ArrayDeque<Runnable> queue = new ArrayDeque<>();
+
+ @Override
+ public synchronized void execute(Runnable command) {
+ queue.add(command);
+ }
+
+ synchronized int size() {
+ return queue.size();
+ }
+
+ /** Remove and return all currently-queued tasks (tasks added later are NOT included). */
+ synchronized List<Runnable> takeAll() {
+ List<Runnable> snapshot = new ArrayList<>(queue);
+ queue.clear();
+ return snapshot;
+ }
+
+ /** Run one task FIFO; returns false if empty. */
+ boolean runOne() {
+ Runnable r;
+ synchronized (this) {
+ r = queue.poll();
+ }
+ if (r == null) {
+ return false;
+ }
+ r.run();
+ return true;
+ }
+
+ @Override
+ public void shutdown() {
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return takeAll();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return false;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return false;
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return true;
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // Operation alphabet (each drives REAL code)
+ // -----------------------------------------------------------------------------------------------
+
+ /** Connect a consumer via the REAL PersistentSubscription.addConsumer path; for Failover with a zero
+ * failover delay this arms the tail-wait read synchronously through the dispatcher's
+ * scheduleReadOnActiveConsumer -> readMoreEntries. */
+ private void connect(Consumer consumerToAdd) throws Exception {
+ subscription.addConsumer(consumerToAdd).get();
+ }
+
+ /**
+ * Disconnect a consumer via the REAL {@code PersistentSubscription.removeConsumer(consumer, false)} path,
+ * so dispatcher removal (which calls {@code cancelPendingRead()}), {@code deactivateCursor()} and
+ * {@code removeWaitingCursor()} all run in the production order. For a durable cursor the subscription
+ * deliberately does not cancel the pending read — see the durable-cursor comment in
+ * {@code PersistentSubscription#removeConsumer}.
+ */
+ private void disconnect(Consumer consumerToRemove) throws Exception {
+ subscription.removeConsumer(consumerToRemove, false);
+ }
+
+ private Position publish() throws Exception {
+ return ledger.addEntry(("m" + msgCounter.incrementAndGet()).getBytes());
+ }
+
+ private void ackUpTo(Position p) throws Exception {
+ cursor.markDelete(p);
+ }
+
+ /** Block (with an explicit timeout) until the dispatcher board holds at least {@code n} tasks. */
+ private void awaitBoardHasAtLeast(int n) {
+ Awaitility.await("dispatcher board should hold at least " + n + " task(s)")
+ .atMost(Duration.ofSeconds(BOARD_WAIT_SECONDS))
+ .pollInterval(Duration.ofMillis(2))
+ .until(() -> dispatcherExecutor.size() >= n);
+ }
+
+ /**
+ * Drain the dispatcher board until nothing new is produced, with a hard deadline so a regression can
+ * never hang CI. Async read completions posted by the (real) managed-ledger worker pool land on the
+ * board between idle rounds.
+ */
+ private void quiesce() {
+ long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(QUIESCE_TIMEOUT_SECONDS);
+ for (int idleRounds = 0; idleRounds < QUIESCE_IDLE_ROUNDS;) {
+ if (System.nanoTime() > deadlineNanos) {
+ throw new IllegalStateException("quiesce() did not settle within " + QUIESCE_TIMEOUT_SECONDS
+ + "s; board=" + dispatcherExecutor.size() + " pendingReadOps="
+ + cursor.getPendingReadOpsCount());
+ }
+ boolean did = false;
+ while (dispatcherExecutor.runOne()) {
+ did = true;
+ }
+ if (did) {
+ idleRounds = 0;
+ continue;
+ }
+ if (dispatcherExecutor.size() == 0 && cursor.getPendingReadOpsCount() == 0) {
+ idleRounds++;
+ }
+ sleepQuietly();
+ }
+ }
+
+ /**
+ * Keep draining the dispatcher board until {@code position} has been handed to {@code target}'s
+ * {@code sendMessages}, with a hard deadline so a regression fails instead of hanging CI.
+ */
+ private boolean pumpUntilDelivered(Consumer target, Position position) {
+ Delivery expected = Delivery.of(target.consumerName(), position);
+ long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(DELIVERY_TIMEOUT_SECONDS);
+ do {
+ while (dispatcherExecutor.runOne()) {
+ // A delivery posts follow-up work onto the board; keep draining before re-checking.
+ }
+ if (deliveries.contains(expected)) {
+ return true;
+ }
+ sleepQuietly();
+ } while (System.nanoTime() < deadlineNanos);
+ return false;
+ }
+
+ private static void sleepQuietly() {
+ try {
+ Thread.sleep(3);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // State snapshot + predicates
+ // -----------------------------------------------------------------------------------------------
+
+ /** One entry handed to a consumer by the dispatcher's real delivery tail. */
+ private record Delivery(String consumerName, long ledgerId, long entryId) {
+
+ static Delivery of(String consumerName, Position position) {
+ return new Delivery(consumerName, position.getLedgerId(), position.getEntryId());
+ }
+
+ @Override
+ public String toString() {
+ return consumerName + "@" + ledgerId + ":" + entryId;
+ }
+ }
+
+ /**
+ * A snapshot of the four coupled state variables. {@code inQueue} is derived from
+ * {@code ledger.getWaitingCursorsCount()}; the single-cursor fixture makes that ledger-wide count a
+ * per-cursor membership proxy (0 ⇔ this cursor is not queued).
+ */
+ private record StuckState(boolean armed, boolean inQueue, int pendingReadOps, boolean havePendingRead) {
+
+ /** The absorbing orphan state: armed, not queued, no in-flight read, and havePendingRead cleared. */
+ boolean stranded() {
+ return armed && !inQueue && pendingReadOps == 0 && !havePendingRead;
+ }
+
+ /** The benign seed: armed with havePendingRead cleared, but still queued (so not yet stranded). */
+ boolean seed() {
+ return armed && inQueue && !havePendingRead;
+ }
+
+ @Override
+ public String toString() {
+ return "armed=" + armed + " inQueue=" + inQueue + " pendingReadOps=" + pendingReadOps
+ + " havePendingRead=" + havePendingRead;
+ }
+ }
+
+ private StuckState snapshot() {
+ return new StuckState(cursor.hasPendingReadRequest(), ledger.getWaitingCursorsCount() >= 1,
+ cursor.getPendingReadOpsCount(), dispatcher.havePendingRead);
+ }
+
+ /**
+ * Tail shared by both schedules, asserting the externally observable contract of #26164: after the last
+ * consumer disconnects and a fresh consumer reconnects to the same durable subscription, a message
+ * published afterwards must actually be delivered to that consumer. On buggy master the primary
+ * schedule leaves the cursor in the absorbing orphan state, the reconnect's re-arm CAS-fails and the
+ * message never reaches {@code sendMessages}.
+ */
+ private void assertSubscriptionDeliversAfterReconnect(String schedule, StuckState preDisconnect)
+ throws Exception {
+ disconnect(consumer);
+ quiesce();
+ StuckState postDisconnect = snapshot();
+ log.info("post-disconnect quiescent state state={}", postDisconnect);
+
+ // A client reconnect: a brand new consumer takes over the (still durable) subscription.
+ Consumer reconnected = newMockConsumer("c2");
+ connect(reconnected);
+ quiesce();
+ StuckState postReconnect = snapshot();
+ log.info("post-reconnect quiescent state state={}", postReconnect);
+
+ Position m2 = publish();
+ boolean deliveredM2 = pumpUntilDelivered(reconnected, m2);
+ quiesce();
+ StuckState afterWake = snapshot();
+ log.info("post-wake-publish state state={} delivered={}", afterWake, deliveredM2);
+
+ String states = "schedule=[" + schedule + "] pre-disconnect=[" + preDisconnect + "] post-disconnect=["
+ + postDisconnect + "] post-reconnect=[" + postReconnect + "] after-wake-publish=[" + afterWake
+ + "] deliveries=" + deliveries;
+
+ // Primary, behavioural: the reconnected consumer must actually receive the new message.
+ Assert.assertTrue(deliveredM2,
+ "BUG REPRODUCED: after the last consumer disconnected and a new one reconnected, the message "
+ + "published next (" + m2 + ") never reached the reconnected consumer's sendMessages within "
+ + DELIVERY_TIMEOUT_SECONDS + "s; the Failover subscription is permanently stuck. " + states);
+ Assert.assertTrue(cursor.getReadPosition().compareTo(m2) > 0,
+ "the cursor read position must have advanced past the delivered message " + m2 + ", but it is "
+ + cursor.getReadPosition() + ". " + states);
+
+ // Secondary, diagnostic: name the internal tuple that causes the stall, so a failure points at it.
+ Assert.assertFalse(afterWake.stranded(),
+ "the cursor must not be left in the absorbing orphan state (armed waitingReadOp, absent from "
+ + "ManagedLedger.waitingCursors, no in-flight read, havePendingRead=false). " + states);
+ }
+
+ /** Stage the shared precondition: an armed tail-wait read plus a read completion pending on the board. */
+ private void stageInFlightReadCompletion() throws Exception {
+ // Arm the tail-wait read: addConsumer -> scheduleReadOnActiveConsumer -> readMoreEntries.
+ connect(consumer);
+ Assert.assertTrue(cursor.hasPendingReadRequest(), "the tail-wait read op should be armed after addConsumer");
+ Assert.assertEquals(ledger.getWaitingCursorsCount(), 1, "the cursor should be registered in waitingCursors");
+ Assert.assertTrue(dispatcher.havePendingRead, "havePendingRead should be true while the tail read is armed");
+
+ // Publish m1: notifyCursors -> notifyEntriesAvailable dispatches the armed op's read, advancing the
+ // read position past m1 and posting readEntriesComplete onto the board. The wait op is now consumed,
+ // so hasPendingReadRequest() is false, but havePendingRead is still true.
+ Position m1 = publish();
+ awaitBoardHasAtLeast(1);
+ Assert.assertFalse(cursor.hasPendingReadRequest(),
+ "the armed op should have been consumed by notifyEntriesAvailable");
+
+ // Telescoped ack: mark-delete m1 before its already-dispatched read completes. This lands the
+ // redeliver's subsequent rewind at the tail with no backlog, so the re-arm is a pure in-memory
+ // tail-wait and no BookKeeper read escapes the ManualExecutor to race quiesce(). The staged state
+ // (mark-delete at tail + an already-dispatched read whose completion is still pending) is
+ // production-reachable via a redeliver->ack->redeliver sequence (see the class Javadoc).
+ ackUpTo(m1);
+
+ // Redeliver: posts internalRedeliver onto the board, behind the still-pending read completion.
+ dispatcher.redeliverUnacknowledgedMessages(consumer, 1L);
+ awaitBoardHasAtLeast(2);
+ }
+
+ /**
+ * The board now holds exactly two tasks from two different producers:
+ * <ol start="0">
+ * <li>{@code readEntriesComplete} — posted by the managed-ledger/BookKeeper completion chain
+ * ({@code whenCompleteAsync});</li>
+ * <li>{@code internalRedeliver} — posted by {@code redeliverUnacknowledgedMessages} (the
+ * client-command thread).</li>
+ * </ol>
+ */
+ private List<Runnable> takeTwoRacingTasks() {
+ List<Runnable> board = dispatcherExecutor.takeAll();
+ Assert.assertEquals(board.size(), 2,
+ "the board must hold exactly two tasks from two producers: the read completion "
+ + "(posted by the managed-ledger/BookKeeper completion chain) and internalRedeliver "
+ + "(posted by the client-command thread)");
+ return board;
+ }
+
+ // ===============================================================================================
+ // (a) PRIMARY reproduction: a redeliver races the completion of an already-dispatched read. m1's ack is
+ // telescoped (applied before that read completes) so the redeliver's rewind lands at the tail and the
+ // re-arm is a pure in-memory tail-wait -- fully deterministic. See the class Javadoc fidelity notes.
+ // ===============================================================================================
+
+ @Test(groups = "broker")
+ public void testFailoverConsumerStuckWhenRedeliverRacesInFlightReadCompletion() throws Exception {
+ buildFixture("sac-stuck-read-race");
+ stageInFlightReadCompletion();
+ List<Runnable> board = takeTwoRacingTasks();
+
+ // Critical interleaving: run the redeliver's re-arm BEFORE the stale read completion. internalRedeliver
+ // clears havePendingRead and arms a fresh read; the stale completion then clears havePendingRead again
+ // while the fresh op stays armed -> the seed state.
+ board.get(1).run(); // internalRedeliver
+ board.get(0).run(); // stale readEntriesComplete
+ quiesce();
+
+ // Observation only (NOT asserted): the race mints the seed (armed op + havePendingRead=false, still
+ // queued). The seed is deliberately not asserted -- a completion-side fix (a read-generation guard on
+ // havePendingRead) would prevent the seed from ever forming, so binding on it would couple this test to
+ // one fix strategy. The binding assertions are the behavioural ones after the disconnect + reconnect.
+ StuckState seed = snapshot();
+ log.info("post-redeliver-race quiescent state (observation only) state={} seedPreconditionHeld={}",
+ seed, seed.seed());
+
+ assertSubscriptionDeliversAfterReconnect("redeliver, then stale completion", seed);
+ }
+
+ // ===============================================================================================
+ // (b) NEGATIVE CONTROL: the same setup but the benign FIFO order [stale completion, then redeliver].
+ // The invariant holds end to end; this PASSES on master.
+ // ===============================================================================================
+
+ @Test(groups = "broker")
+ public void testRedeliverAfterReadCompletionDoesNotStrandCursor() throws Exception {
+ buildFixture("sac-stuck-read-fifo");
+ stageInFlightReadCompletion();
+ List<Runnable> board = takeTwoRacingTasks();
+
+ // Benign FIFO order: the stale completion runs FIRST (clearing havePendingRead), then the redeliver
+ // re-arms and sets havePendingRead=true. The armed op and havePendingRead stay coupled, so no seed
+ // forms and the later disconnect can cancel the op cleanly.
+ board.get(0).run(); // stale readEntriesComplete
+ board.get(1).run(); // internalRedeliver
+ quiesce();
+
+ StuckState healthy = snapshot();
+ log.info("post-FIFO-race quiescent state state={}", healthy);
+ Assert.assertFalse(healthy.stranded(), "the benign FIFO order must not strand the cursor: " + healthy);
+
+ // The disconnect now finds havePendingRead=true, so cancelPendingRead actually cancels the armed op:
+ // no orphan is created, the reconnect re-arms cleanly and the next publish is delivered.
+ assertSubscriptionDeliversAfterReconnect("stale completion, then redeliver (negative control)", healthy);
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
index 8eaaf922c9c7..f24c1bb4d86c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java
@@ -102,7 +102,7 @@ public class PersistentDispatcherSingleActiveConsumerTest extends SharedPulsarBa
Mockito.doAnswer(inv -> {
callReadEntriesFailed.getAndIncrement();
return inv.callRealMethod();
- }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any());
+ }).when(dispatcher).readEntriesFailed(Mockito.any(), Mockito.any(), Mockito.anyLong());
Mockito.doReturn(false).when(cursor).isClosed();
(pulsar) 03/11: [fix][broker] Trigger max read position callback for messages published during transaction buffer recovery (#26234)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d7a7a6881fb7dfcd13634bc174bfc60b8f4650eb
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 23 19:15:23 2026 +0300
[fix][broker] Trigger max read position callback for messages published during transaction buffer recovery (#26234)
(cherry picked from commit 520b0daa3f632e283930624bf29e7a5b4ed45828)
---
.../buffer/impl/TopicTransactionBuffer.java | 21 ++-
.../impl/TopicTransactionBufferRecoveryTest.java | 144 +++++++++++++++++++++
2 files changed, 163 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index 7ae311a54639..aad9862cf4fa 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -150,7 +150,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
public void recoverComplete() {
synchronized (TopicTransactionBuffer.this) {
if (ongoingTxns.isEmpty()) {
- maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+ updateMaxReadPositionAfterRecovery();
}
if (!changeToReadyState()) {
log.error("[{}]Transaction buffer recover fail, current state: {}",
@@ -171,7 +171,7 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
@Override
public void noNeedToRecover() {
synchronized (TopicTransactionBuffer.this) {
- maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+ updateMaxReadPositionAfterRecovery();
if (!changeToNoSnapshotState()) {
log.error("[{}]Transaction buffer recover fail", topic.getName());
} else {
@@ -627,6 +627,23 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
}
}
+ /**
+ * Advance the max read position to the last confirmed entry when recovery finishes. While the transaction
+ * buffer is recovering, {@link #syncMaxReadPositionForNormalPublish(Position, boolean)} ignores publishes, so
+ * messages published during recovery would otherwise never trigger the maxReadPositionMovedForward callback.
+ * {@link PersistentTopic} uses that callback to maintain lastMaxReadPositionMovedForwardTimestamp, which
+ * ReplicatedSubscriptionsController relies on to detect new data when deciding whether to start a snapshot.
+ * Must be called while synchronized on this transaction buffer.
+ */
+ private void updateMaxReadPositionAfterRecovery() {
+ Position preMaxReadPosition = this.maxReadPosition;
+ this.maxReadPosition = topic.getManagedLedger().getLastConfirmedEntry();
+ if (this.maxReadPosition != null
+ && (preMaxReadPosition == null || preMaxReadPosition.compareTo(this.maxReadPosition) < 0)) {
+ this.maxReadPositionCallBack.maxReadPositionMovedForward(preMaxReadPosition, this.maxReadPosition);
+ }
+ }
+
@Override
public CompletableFuture<Void> purgeTxns(List<Long> dataLedgers) {
return null;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
new file mode 100644
index 000000000000..101f69b263f8
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBufferRecoveryTest.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.transaction.buffer.impl;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import lombok.Cleanup;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.PositionFactory;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
+import org.apache.pulsar.broker.transaction.buffer.TransactionBufferProvider;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/**
+ * Tests for {@link TopicTransactionBuffer} recovery completion behavior.
+ */
+@Test(groups = "broker")
+public class TopicTransactionBufferRecoveryTest extends ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setTransactionCoordinatorEnabled(true);
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @DataProvider(name = "snapshotExists")
+ public Object[][] snapshotExists() {
+ return new Object[][] { { false }, { true } };
+ }
+
+ /**
+ * While the transaction buffer is recovering, normal publishes don't move the max read position, so they
+ * don't update the topic's lastMaxReadPositionMovedForwardTimestamp either. When recovery completes, the
+ * transaction buffer must account for the messages published during recovery and trigger the
+ * maxReadPositionMovedForward callback; otherwise ReplicatedSubscriptionsController would consider the topic
+ * to have no new data and never start a subscription snapshot until further traffic arrives.
+ */
+ @Test(dataProvider = "snapshotExists")
+ public void testMaxReadPositionMovedForwardForMessagesPublishedDuringRecovery(boolean snapshotExists)
+ throws Exception {
+ String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp-tb-recovery");
+ CompletableFuture<Position> recoverFuture = new CompletableFuture<>();
+ TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+ pulsar.setTransactionBufferProvider(originTopic -> {
+ AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ when(processor.recoverFromSnapshot()).thenReturn(recoverFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ return new TopicTransactionBuffer(
+ (PersistentTopic) originTopic, processor, AbortedTxnProcessor.SnapshotType.Single);
+ });
+ try {
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(tpName).create();
+ // The transaction buffer is stuck in the recovering state until recoverFuture completes, so these
+ // messages are published while it is still recovering.
+ for (int i = 0; i < 3; i++) {
+ producer.send(("msg-" + i).getBytes(StandardCharsets.UTF_8));
+ }
+ PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopicIfExists(tpName).get().orElseThrow();
+ assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+ "Publishes during recovery are not expected to move the max read position forward");
+
+ // Let the recovery finish, either replaying from a snapshot position or without any snapshot.
+ recoverFuture.complete(snapshotExists ? PositionFactory.EARLIEST : null);
+
+ Awaitility.await().untilAsserted(() -> {
+ assertTrue(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp() > 0,
+ "Completed recovery should move the max read position forward for the messages"
+ + " published during recovery");
+ assertEquals(persistentTopic.getTransactionBuffer().getMaxReadPosition(),
+ persistentTopic.getManagedLedger().getLastConfirmedEntry());
+ });
+ } finally {
+ pulsar.setTransactionBufferProvider(originalProvider);
+ }
+ }
+
+ /**
+ * When nothing is published while the transaction buffer recovers, completing the recovery must not trigger
+ * the maxReadPositionMovedForward callback, so an idle topic isn't mistaken for one with new data.
+ */
+ @Test
+ public void testMaxReadPositionNotMovedForwardWhenNothingPublishedDuringRecovery() throws Exception {
+ String tpName = BrokerTestUtil.newUniqueName("persistent://public/default/tp-tb-recovery-idle");
+ CompletableFuture<Position> recoverFuture = new CompletableFuture<>();
+ TransactionBufferProvider originalProvider = pulsar.getTransactionBufferProvider();
+ pulsar.setTransactionBufferProvider(originTopic -> {
+ AbortedTxnProcessor processor = mock(AbortedTxnProcessor.class);
+ when(processor.recoverFromSnapshot()).thenReturn(recoverFuture);
+ when(processor.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
+ return new TopicTransactionBuffer(
+ (PersistentTopic) originTopic, processor, AbortedTxnProcessor.SnapshotType.Single);
+ });
+ try {
+ PersistentTopic persistentTopic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopic(tpName, true).get().orElseThrow();
+ recoverFuture.complete(null);
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(persistentTopic.getTransactionBuffer().getMaxReadPosition(),
+ persistentTopic.getManagedLedger().getLastConfirmedEntry()));
+ assertEquals(persistentTopic.getLastMaxReadPositionMovedForwardTimestamp(), 0L,
+ "Recovery of an idle topic should not move the max read position forward");
+ } finally {
+ pulsar.setTransactionBufferProvider(originalProvider);
+ }
+ }
+}
(pulsar) 06/11: [fix][ml] Preserve ledger entries/size when transformLedgerInfo callback completes after a concurrent close (#26228)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1d9dadaba53b47b85c1d69944ca17edafe0bd46c
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Fri Jul 24 18:01:08 2026 +0800
[fix][ml] Preserve ledger entries/size when transformLedgerInfo callback completes after a concurrent close (#26228)
(cherry picked from commit 1aa37ff839520e5c0a9e5da58d41dfbd0732546b)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 18 +-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 204 +++++++++++++++++++++
2 files changed, 220 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 0d4530e11e86..9c77e4a9b556 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
@@ -3880,11 +3880,25 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
final HashMap<Long, LedgerInfo> newLedgers = new HashMap<>(ledgers);
newLedgers.put(ledgerId, newInfo);
store.asyncUpdateLedgerIds(name, buildManagedLedgerInfo(newLedgers), ledgersStat,
- new MetaStoreCallback<Void>() {
+ new MetaStoreCallback<>() {
@Override
public void operationComplete(Void result, Stat stat) {
ledgersStat = stat;
- ledgers.put(ledgerId, newInfo);
+ ledgers.computeIfPresent(ledgerId, (id, existing) -> {
+ LedgerInfo merged = new LedgerInfo();
+ merged.copyFrom(newInfo);
+ merged.clearEntries().clearSize().clearTimestamp();
+ if (existing.hasEntries()) {
+ merged.setEntries(existing.getEntries());
+ }
+ if (existing.hasSize()) {
+ merged.setSize(existing.getSize());
+ }
+ if (existing.hasTimestamp()) {
+ merged.setTimestamp(existing.getTimestamp());
+ }
+ return merged;
+ });
unlockingPromise.complete(null);
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index b8ae552ff6ab..cc9b97f9bb8a 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -4711,6 +4711,210 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
Assert.assertEquals(ml.getLedgersInfo().get(lastLedger).getPropertiesCount(), 0);
}
+ /**
+ * Verifies that a ledger-property write whose metadata-store callback completes AFTER the ledger
+ * has been closed does not regress the closed ledger's entries.
+ *
+ * <p>The transform callback in {@code tryTransformLedgerInfo} previously did a blind
+ * {@code ledgers.put(ledgerId, newInfo)}, where {@code newInfo} carried the entry count captured at
+ * transform time (before the close). If the ledger filled and {@code ledgerClosed} updated the
+ * in-memory entry count in the meantime, the callback overwrote it with the stale pre-close value.
+ * The fix merges instead, keeping the transform's properties but taking entries/size/timestamp
+ * from the current in-memory value.
+ */
+ @Test
+ public void testLedgerPropertyWriteDoesNotRegressEntriesAfterConcurrentClose() throws Exception {
+ final String mlName = "testLedgerPropertyWriteDoesNotRegressEntriesAfterConcurrentClose";
+ final String mlPath = "/managed-ledgers/" + mlName;
+ final String key = "k";
+ final String value = "v";
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(2);
+
+ // Gate that delays completion of the property write's managed-ledger PUT (and therefore its
+ // transform callback) until the ledger has been closed. We must NOT block inside the spy:
+ // asyncUpdateLedgerIds runs while the managed-ledger monitor is held, so blocking would
+ // deadlock addEntry. Instead we return a future that completes only after releaseGate.
+ CompletableFuture<Void> releaseGate = new CompletableFuture<>();
+ AtomicBoolean interceptNextPut = new AtomicBoolean(false);
+ CountDownLatch putIntercepted = new CountDownLatch(1);
+
+ FaultInjectionMetadataStore spyStore = spy(metadataStore);
+ doAnswer(inv -> {
+ if (mlPath.equals(inv.getArgument(0)) && interceptNextPut.compareAndSet(true, false)) {
+ putIntercepted.countDown();
+ CompletableFuture<Stat> real = (CompletableFuture<Stat>) inv.callRealMethod();
+ CompletableFuture<Stat> gated = new CompletableFuture<>();
+ // Forward the real result to `gated` only once `releaseGate` is completed.
+ real.whenComplete((stat, ex) -> releaseGate.whenComplete((ignored, ignoredEx) -> {
+ if (ex != null) {
+ gated.completeExceptionally(ex);
+ } else {
+ gated.complete(stat);
+ }
+ }));
+ return gated;
+ }
+ return inv.callRealMethod();
+ }).when(spyStore).put(eq(mlPath), any(byte[].class), any());
+
+ ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(spyStore, bkc);
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(mlName, config);
+ try {
+ ml.addEntry("e1".getBytes()); // ledger X: 1 entry, not yet full
+ final long x = ml.currentLedger.getId();
+
+ // Kick off the property write; its managed-ledger PUT is intercepted and held open.
+ interceptNextPut.set(true);
+ CompletableFuture<Void> prop = ml.asyncAddLedgerProperty(x, key, value);
+ Assert.assertTrue(putIntercepted.await(5, TimeUnit.SECONDS));
+
+ // Fill ledger X -> ledgerClosed(X) updates the in-memory entries/size/timestamp to the
+ // closed values.
+ ml.addEntry("e2".getBytes());
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> Assert.assertEquals(2L, ml.getLedgersInfo().get(x).getEntries()));
+ long closedSize = ml.getLedgersInfo().get(x).getSize();
+ long closedTimestamp = ml.getLedgersInfo().get(x).getTimestamp();
+
+ // Now let the property write's callback run: it must merge, not overwrite, the closed values.
+ releaseGate.complete(null);
+ prop.get();
+
+ LedgerInfo info = ml.getLedgersInfo().get(x);
+ // Without the merge fix entries regressed to 0 (captured before the close); size/timestamp
+ // were likewise clobbered by the stale snapshot.
+ Assert.assertEquals(2L, info.getEntries());
+ Assert.assertEquals(closedSize, info.getSize());
+ Assert.assertEquals(closedTimestamp, info.getTimestamp());
+ Assert.assertEquals(value, ml.asyncGetLedgerProperty(x, key).get());
+ } finally {
+ releaseGate.complete(null); // never leave the gated PUT dangling
+ ml.close();
+ factory.shutdown();
+ }
+ }
+
+ /**
+ * Verifies the no-concurrent-close path: when a property write completes on a still-open current
+ * ledger, the transform's merge must preserve the unset (not present) lightproto optional fields
+ * {@code entries}/{@code size}/{@code timestamp} rather than converting them into
+ * explicitly-present zeros.
+ */
+ @Test
+ public void testLedgerPropertyWritePreservesUnsetFieldPresenceOnOpenLedger() throws Exception {
+ final String mlName = "testLedgerPropertyWritePreservesUnsetFieldPresenceOnOpenLedger";
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(2);
+
+ ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(mlName, config);
+ try {
+ ml.addEntry("e1".getBytes()); // current ledger X is still open: entries/size unset
+ final long x = ml.currentLedger.getId();
+
+ // No concurrent close: the property write completes while the ledger is still open.
+ ml.asyncAddLedgerProperty(x, "k", "v").get();
+
+ LedgerInfo info = ml.getLedgersInfo().get(x);
+ // A newly created current ledger has no entries/size yet; the merge must not convert
+ // those unset optional fields into explicitly-present zeros.
+ Assert.assertFalse(info.hasEntries());
+ Assert.assertFalse(info.hasSize());
+ Assert.assertEquals("v", ml.asyncGetLedgerProperty(x, "k").get());
+ } finally {
+ ml.close();
+ factory.shutdown();
+ }
+ }
+
+ /**
+ * Verifies that the corrected entries/size/timestamp survive a restart: after the
+ * concurrent-close merge fixes the in-memory values, a subsequent ledger-list persist durably
+ * records them, and reopening the managed ledger reads them back.
+ */
+ @Test
+ public void testLedgerPropertyWriteCorrectedValuesSurviveReopen() throws Exception {
+ final String mlName = "testLedgerPropertyWriteCorrectedValuesSurviveReopen";
+ final String mlPath = "/managed-ledgers/" + mlName;
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(2);
+
+ CompletableFuture<Void> releaseGate = new CompletableFuture<>();
+ AtomicBoolean interceptNextPut = new AtomicBoolean(false);
+ CountDownLatch putIntercepted = new CountDownLatch(1);
+
+ FaultInjectionMetadataStore spyStore = spy(metadataStore);
+ doAnswer(inv -> {
+ if (mlPath.equals(inv.getArgument(0)) && interceptNextPut.compareAndSet(true, false)) {
+ putIntercepted.countDown();
+ CompletableFuture<Stat> real = (CompletableFuture<Stat>) inv.callRealMethod();
+ CompletableFuture<Stat> gated = new CompletableFuture<>();
+ real.whenComplete((stat, ex) -> releaseGate.whenComplete((ignored, ignoredEx) -> {
+ if (ex != null) {
+ gated.completeExceptionally(ex);
+ } else {
+ gated.complete(stat);
+ }
+ }));
+ return gated;
+ }
+ return inv.callRealMethod();
+ }).when(spyStore).put(eq(mlPath), any(byte[].class), any());
+
+ ManagedLedgerFactoryImpl factory = new ManagedLedgerFactoryImpl(spyStore, bkc);
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) factory.open(mlName, config);
+ final long x;
+ long closedSize;
+ long closedTimestamp;
+ try {
+ ml.addEntry("e1".getBytes()); // ledger X: 1 entry, not yet full
+ x = ml.currentLedger.getId();
+
+ interceptNextPut.set(true);
+ CompletableFuture<Void> prop = ml.asyncAddLedgerProperty(x, "k", "v");
+ Assert.assertTrue(putIntercepted.await(5, TimeUnit.SECONDS));
+
+ // Close X concurrently so the merge is the only thing keeping entries/size/timestamp correct.
+ ml.addEntry("e2".getBytes());
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> Assert.assertEquals(2L, ml.getLedgersInfo().get(x).getEntries()));
+ closedSize = ml.getLedgersInfo().get(x).getSize();
+ closedTimestamp = ml.getLedgersInfo().get(x).getTimestamp();
+
+ releaseGate.complete(null);
+ prop.get();
+
+ // Trigger another ledger-list persist so the corrected values reach the metadata store.
+ ml.addEntry("e3".getBytes());
+ Awaitility.await().atMost(Duration.ofSeconds(5))
+ .untilAsserted(() -> Assert.assertEquals(2L, ml.getLedgersInfo().get(x).getEntries()));
+ } finally {
+ releaseGate.complete(null);
+ ml.close();
+ factory.shutdown();
+ }
+
+ // Reopen from the metadata store and confirm the corrected values survived.
+ ManagedLedgerFactoryImpl factory2 = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ try {
+ ManagedLedgerImpl ml2 = (ManagedLedgerImpl) factory2.open(mlName, config);
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+ LedgerInfo info = ml2.getLedgersInfo().get(x);
+ Assert.assertNotNull(info);
+ Assert.assertEquals(2L, info.getEntries());
+ Assert.assertEquals(closedSize, info.getSize());
+ Assert.assertEquals(closedTimestamp, info.getTimestamp());
+ });
+ ml2.close();
+ } finally {
+ factory2.shutdown();
+ }
+ }
+
/**
* Verifies that ledger trimming respects the persistent cursor position, not just the in-memory position.
*
(pulsar) 02/11: [fix][broker] Check deliverAt before containsMessage in bucket addMessage (#26230)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 242202b1e318ac83813be0bd7027cd13a25414c6
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Thu Jul 23 17:20:44 2026 +0800
[fix][broker] Check deliverAt before containsMessage in bucket addMessage (#26230)
(cherry picked from commit 094f2702833ac0ddf71beae4df2072048861eaf5)
---
.../bucket/BucketDelayedDeliveryTracker.java | 8 ++--
.../bucket/BucketDelayedDeliveryTrackerTest.java | 48 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
index 4b6121fd3e4a..8662867d1828 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java
@@ -367,14 +367,14 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
@Override
public synchronized boolean addMessage(long ledgerId, long entryId, long deliverAt) {
- if (containsMessage(ledgerId, entryId)) {
- return true;
- }
-
if (deliverAt < 0 || deliverAt <= getCutoffTime()) {
return false;
}
+ if (containsMessage(ledgerId, entryId)) {
+ return true;
+ }
+
boolean existBucket = findImmutableBucket(ledgerId).isPresent();
// Create bucket snapshot
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
index 4becc00ab3f3..98f3f271ec12 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java
@@ -45,6 +45,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
+import lombok.Cleanup;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.bookkeeper.mledger.Position;
@@ -154,6 +155,10 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
new BucketDelayedDeliveryTracker(dispatcher, timer, 100000, clock,
true, bucketSnapshotStorage, 20, TimeUnit.HOURS.toMillis(1), 5, 100)
}};
+ case "testExpiredTrackedMessageReturnsFalse", "testRecoverThenExpireAddMessage" -> new Object[][]{{
+ new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock,
+ true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50)
+ }};
default -> new Object[][]{{
new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock,
true, bucketSnapshotStorage, 1000, TimeUnit.MILLISECONDS.toMillis(100), -1, 50)
@@ -185,6 +190,49 @@ public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTes
tracker.close();
}
+ @Test(dataProvider = "delayedTracker")
+ public void testExpiredTrackedMessageReturnsFalse(BucketDelayedDeliveryTracker tracker) {
+ clockTime.set(1000);
+ assertTrue(tracker.addMessage(1, 1, 2000));
+ assertTrue(tracker.containsMessage(1, 1));
+
+ clockTime.set(2500);
+
+ assertFalse(
+ "Expired tracked message should return false so dispatcher delivers immediately",
+ tracker.addMessage(1, 1, 2000));
+
+ tracker.close();
+ }
+
+ @Test(dataProvider = "delayedTracker")
+ public void testRecoverThenExpireAddMessage(BucketDelayedDeliveryTracker tracker) throws Exception {
+ clockTime.set(0);
+ for (int i = 1; i <= 6; i++) {
+ tracker.addMessage(i, i, i * 1000);
+ }
+
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(tracker.getImmutableBuckets().asMapOfRanges().values().stream()
+ .noneMatch(x -> x.merging || !x.getSnapshotCreateFuture().get().isDone())));
+
+ tracker.close();
+
+ clockTime.set(0);
+ @Cleanup
+ BucketDelayedDeliveryTracker tracker2 = new BucketDelayedDeliveryTracker(
+ dispatcher, timer, 100000, clock, true,
+ bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50);
+
+ assertTrue(tracker2.containsMessage(1, 1));
+
+ clockTime.set(10000);
+
+ assertFalse(
+ "Recovered message that is now expired should return false",
+ tracker2.addMessage(1, 1, 1000));
+ }
+
@Test(dataProvider = "delayedTracker", invocationCount = 10)
public void testRecoverSnapshot(BucketDelayedDeliveryTracker tracker) throws Exception {
for (int i = 1; i <= 100; i++) {
(pulsar) 04/11: [improve][misc] Upgrade Jetty to 12.1.11 (#26233)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 83af298f43f7385b7a045aee0db4527b0eca7ad0
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Jul 23 22:15:14 2026 +0300
[improve][misc] Upgrade Jetty to 12.1.11 (#26233)
(cherry picked from commit 4bfb08063e753fa3d9b982b283f8cefe94dc6230)
---
distribution/server/src/assemble/LICENSE.bin.txt | 72 ++++++++++++------------
distribution/shell/src/assemble/LICENSE.bin.txt | 24 ++++----
pom.xml | 2 +-
3 files changed, 49 insertions(+), 49 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index fad9617c9a8c..c6ccabab0cf2 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -393,43 +393,43 @@ The Apache Software License, Version 2.0
- org.asynchttpclient-async-http-client-2.15.0.jar
- org.asynchttpclient-async-http-client-netty-utils-2.15.0.jar
* Jetty
- - org.eclipse.jetty-jetty-alpn-client-12.1.10.jar
- - org.eclipse.jetty-jetty-alpn-conscrypt-server-12.1.10.jar
- - org.eclipse.jetty-jetty-alpn-server-12.1.10.jar
- - org.eclipse.jetty-jetty-annotations-12.1.10.jar
- - org.eclipse.jetty-jetty-client-12.1.10.jar
- - org.eclipse.jetty-jetty-http-12.1.10.jar
- - org.eclipse.jetty-jetty-io-12.1.10.jar
- - org.eclipse.jetty-jetty-jndi-12.1.10.jar
- - org.eclipse.jetty-jetty-plus-12.1.10.jar
- - org.eclipse.jetty-jetty-security-12.1.10.jar
- - org.eclipse.jetty-jetty-server-12.1.10.jar
- - org.eclipse.jetty-jetty-session-12.1.10.jar
- - org.eclipse.jetty-jetty-util-12.1.10.jar
- - org.eclipse.jetty-jetty-xml-12.1.10.jar
- - org.eclipse.jetty.compression-jetty-compression-common-12.1.10.jar
- - org.eclipse.jetty.compression-jetty-compression-gzip-12.1.10.jar
- - org.eclipse.jetty.compression-jetty-compression-server-12.1.10.jar
- - org.eclipse.jetty.ee-jetty-ee-webapp-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-annotations-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-nested-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-plus-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-proxy-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-security-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-servlet-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-servlets-12.1.10.jar
- - org.eclipse.jetty.ee8-jetty-ee8-webapp-12.1.10.jar
- - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-api-12.1.10.jar
- - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-common-12.1.10.jar
- - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-server-12.1.10.jar
- - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-servlet-12.1.10.jar
+ - org.eclipse.jetty-jetty-alpn-client-12.1.11.jar
+ - org.eclipse.jetty-jetty-alpn-conscrypt-server-12.1.11.jar
+ - org.eclipse.jetty-jetty-alpn-server-12.1.11.jar
+ - org.eclipse.jetty-jetty-annotations-12.1.11.jar
+ - org.eclipse.jetty-jetty-client-12.1.11.jar
+ - org.eclipse.jetty-jetty-http-12.1.11.jar
+ - org.eclipse.jetty-jetty-io-12.1.11.jar
+ - org.eclipse.jetty-jetty-jndi-12.1.11.jar
+ - org.eclipse.jetty-jetty-plus-12.1.11.jar
+ - org.eclipse.jetty-jetty-security-12.1.11.jar
+ - org.eclipse.jetty-jetty-server-12.1.11.jar
+ - org.eclipse.jetty-jetty-session-12.1.11.jar
+ - org.eclipse.jetty-jetty-util-12.1.11.jar
+ - org.eclipse.jetty-jetty-xml-12.1.11.jar
+ - org.eclipse.jetty.compression-jetty-compression-common-12.1.11.jar
+ - org.eclipse.jetty.compression-jetty-compression-gzip-12.1.11.jar
+ - org.eclipse.jetty.compression-jetty-compression-server-12.1.11.jar
+ - org.eclipse.jetty.ee-jetty-ee-webapp-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-annotations-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-nested-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-plus-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-proxy-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-security-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-servlet-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-servlets-12.1.11.jar
+ - org.eclipse.jetty.ee8-jetty-ee8-webapp-12.1.11.jar
+ - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-api-12.1.11.jar
+ - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-common-12.1.11.jar
+ - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-jetty-server-12.1.11.jar
+ - org.eclipse.jetty.ee8.websocket-jetty-ee8-websocket-servlet-12.1.11.jar
- org.eclipse.jetty.toolchain-jetty-servlet-api-4.0.9.jar
- - org.eclipse.jetty.websocket-jetty-websocket-core-client-12.1.10.jar
- - org.eclipse.jetty.websocket-jetty-websocket-core-common-12.1.10.jar
- - org.eclipse.jetty.websocket-jetty-websocket-core-server-12.1.10.jar
- - org.eclipse.jetty.websocket-jetty-websocket-jetty-api-12.1.10.jar
- - org.eclipse.jetty.websocket-jetty-websocket-jetty-client-12.1.10.jar
- - org.eclipse.jetty.websocket-jetty-websocket-jetty-common-12.1.10.jar
+ - org.eclipse.jetty.websocket-jetty-websocket-core-client-12.1.11.jar
+ - org.eclipse.jetty.websocket-jetty-websocket-core-common-12.1.11.jar
+ - org.eclipse.jetty.websocket-jetty-websocket-core-server-12.1.11.jar
+ - org.eclipse.jetty.websocket-jetty-websocket-jetty-api-12.1.11.jar
+ - org.eclipse.jetty.websocket-jetty-websocket-jetty-client-12.1.11.jar
+ - org.eclipse.jetty.websocket-jetty-websocket-jetty-common-12.1.11.jar
* SnakeYaml -- org.yaml-snakeyaml-2.0.jar
* RocksDB - org.rocksdb-rocksdbjni-7.9.2.jar
* Google Error Prone Annotations - com.google.errorprone-error_prone_annotations-2.45.0.jar
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index eddd1b0fad58..b9bb87f35587 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -401,18 +401,18 @@ The Apache Software License, Version 2.0
- async-http-client-2.15.0.jar
- async-http-client-netty-utils-2.15.0.jar
* Jetty
- - jetty-alpn-client-12.1.10.jar
- - jetty-client-12.1.10.jar
- - jetty-compression-common-12.1.10.jar
- - jetty-compression-gzip-12.1.10.jar
- - jetty-http-12.1.10.jar
- - jetty-io-12.1.10.jar
- - jetty-util-12.1.10.jar
- - jetty-websocket-core-client-12.1.10.jar
- - jetty-websocket-core-common-12.1.10.jar
- - jetty-websocket-jetty-api-12.1.10.jar
- - jetty-websocket-jetty-client-12.1.10.jar
- - jetty-websocket-jetty-common-12.1.10.jar
+ - jetty-alpn-client-12.1.11.jar
+ - jetty-client-12.1.11.jar
+ - jetty-compression-common-12.1.11.jar
+ - jetty-compression-gzip-12.1.11.jar
+ - jetty-http-12.1.11.jar
+ - jetty-io-12.1.11.jar
+ - jetty-util-12.1.11.jar
+ - jetty-websocket-core-client-12.1.11.jar
+ - jetty-websocket-core-common-12.1.11.jar
+ - jetty-websocket-jetty-api-12.1.11.jar
+ - jetty-websocket-jetty-client-12.1.11.jar
+ - jetty-websocket-jetty-common-12.1.11.jar
* SnakeYaml -- snakeyaml-2.0.jar
* Google Error Prone Annotations - error_prone_annotations-2.45.0.jar
* Javassist -- javassist-3.25.0-GA.jar
diff --git a/pom.xml b/pom.xml
index 61de2e58ee35..0ce4eb06bf9a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -191,7 +191,7 @@ flexible messaging model and an intuitive client API.</description>
<curator.version>5.7.1</curator.version>
<netty.version>4.1.136.Final</netty.version>
<netty-iouring.version>0.0.26.Final</netty-iouring.version>
- <jetty.version>12.1.10</jetty.version>
+ <jetty.version>12.1.11</jetty.version>
<!-- Jetty 9 is used in tiered-storage/file-system tests and pulsar-io/alluxio -->
<jetty9.version>9.4.58.v20250814</jetty9.version>
<conscrypt.version>2.5.2</conscrypt.version>
(pulsar) 09/11: [fix][broker] Fix Key_Shared delivery stall when look-ahead triggers at the end of the topic (#26236)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f066cfe8a09e991ee153d2fb90f1148e2a4386da
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Jul 25 13:57:02 2026 +0300
[fix][broker] Fix Key_Shared delivery stall when look-ahead triggers at the end of the topic (#26236)
(cherry picked from commit 37ceb98fb0a2feecd8eb295207edcee30049c0ba)
---
...istentStickyKeyDispatcherMultipleConsumers.java | 13 ++-
...ntStickyKeyDispatcherMultipleConsumersTest.java | 99 ++++++++++++++++++++++
2 files changed, 111 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 86c4329521b3..3e0354a7b394 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -315,13 +315,24 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
acquirePermitsForDeliveredMessages(topic, cursor, totalEntries, totalMessagesSent, totalBytesSent);
// trigger read more messages if necessary
- if (triggerLookAhead.booleanValue()) {
+ if (triggerLookAhead.booleanValue() && (allowOutOfOrderDelivery || cursor.hasMoreEntries())) {
// When all messages get filtered and no messages are sent, we should read more entries, "look ahead"
// so that a possible next batch of messages might contain messages that can be dispatched.
// This is done only when there's a consumer with available permits, and it's not able to make progress
// because of blocked hashes. Without this rule we would be looking ahead in the stream while the
// new consumers are not ready to accept the new messages,
// therefore would be most likely only increase the distance between read-position and mark-delete position.
+ // When ordered delivery is required, look-ahead is engaged only when the cursor has more entries.
+ // Otherwise the next readMoreEntries call would skip replaying the replay queue and pulling due messages
+ // from the delayed delivery tracker, and instead issue a normal read that waits at the end of the topic
+ // for new entries. That would leave deliverable messages stuck in the replay queue or the delayed
+ // delivery tracker until an unrelated event (such as a consumer flow request) triggers another read,
+ // stalling dispatch (issue #21554).
+ // When out-of-order delivery is allowed, look-ahead is engaged unconditionally, as before. In that mode
+ // the replay queue doesn't track sticky key hashes, so the replay position filter cannot exclude
+ // messages for consumers without available permits, and each replay would re-read and discard the same
+ // undispatchable messages. Ending the cycle with a look-ahead attempt (that waits at the end of the
+ // topic when there is nothing to read) prevents such repeated read-and-discard loops.
skipNextReplayToTriggerLookAhead = true;
// skip backoff delay before reading ahead in the "look ahead" mode to prevent any additional latency
// only skip the delay if there are more entries to read
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
index 26e16616d1ad..50007b5c8619 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersTest.java
@@ -463,6 +463,105 @@ public class PersistentStickyKeyDispatcherMultipleConsumersTest {
allEntries.forEach(Entry::release);
}
+ /**
+ * Reproduces the dispatch stall behind the flaky
+ * KeySharedSubscriptionTest.testContinueDispatchMessagesWhenMessageDelayed (issue #21554).
+ *
+ * When a replay read gets fully discarded (for example because the target consumer ran out of permits while
+ * the read was in flight) and the cursor has no more entries, engaging the "look ahead" mode made the follow-up
+ * read skip the replay queue and issue a normal read that waits at the end of the topic for new entries. Messages
+ * in the replay queue for consumers with available permits were then stuck until an unrelated event, such as a
+ * consumer flow request, triggered another read.
+ */
+ @Test(timeOut = 30000)
+ public void testLookAheadNotEngagedWhenCursorHasNoMoreEntries() throws Exception {
+ persistentDispatcher.close();
+
+ // the mocked executor doesn't support schedule(), so run the rescheduled read directly
+ persistentDispatcher = new PersistentStickyKeyDispatcherMultipleConsumers(
+ topicMock, cursorMock, subscriptionMock, configMock,
+ new KeySharedMeta().setKeySharedMode(KeySharedMode.AUTO_SPLIT)) {
+ @Override
+ protected void reScheduleReadInMs(long readAfterMs) {
+ orderedExecutor.execute(this::readMoreEntries);
+ }
+ };
+
+ // consumer1 has available permits at all times
+ persistentDispatcher.addConsumer(consumerMock).join();
+
+ // the slow consumer initially has 2 permits
+ final Consumer slowConsumerMock = createMockConsumer();
+ doReturn("consumer2").when(slowConsumerMock).consumerName();
+ doReturn(true).when(slowConsumerMock).isWritable();
+ AtomicInteger slowConsumerAvailablePermits = new AtomicInteger(2);
+ doAnswer(invocation -> slowConsumerAvailablePermits.get()).when(slowConsumerMock).getAvailablePermits();
+ persistentDispatcher.addConsumer(slowConsumerMock).join();
+
+ StickyKeyConsumerSelector selector = persistentDispatcher.getSelector();
+ String keyForConsumer1 = generateKeyForConsumer(selector, consumerMock);
+ String keyForSlowConsumer = generateKeyForConsumer(selector, slowConsumerMock);
+
+ final Entry entry1 = createEntry(1, 1, "message1", 1, keyForSlowConsumer);
+ final Entry entry2 = createEntry(1, 2, "message2", 2, keyForSlowConsumer);
+ final Entry entry3 = createEntry(1, 3, "message3", 3, keyForConsumer1);
+ final List<Entry> allEntries = List.of(entry1, entry2, entry3);
+
+ // the cursor has no more entries; a normal read would wait for new entries without completing
+ doReturn(false).when(cursorMock).hasMoreEntries();
+ doAnswer(invocationOnMock -> null)
+ .when(cursorMock).asyncReadEntriesWithSkipOrWait(anyInt(), anyLong(), any(), any(), any(), any());
+
+ // Mock Cursor#asyncReplayEntries. While the first replay read is in flight, the slow consumer runs out of
+ // permits and a message for consumer1 becomes replayable. This mirrors the delayed delivery tracker feeding
+ // messages to the replay queue while dispatching is in progress.
+ doAnswer(invocationOnMock -> {
+ Set<Position> positionsArg = invocationOnMock.getArgument(0);
+ Set<Position> positions = new TreeSet<>(positionsArg);
+ if (!positions.contains(entry3.getPosition())) {
+ slowConsumerAvailablePermits.set(0);
+ // add extra retain since addEntryToReplay will release it
+ ((EntryImpl) entry3).retain();
+ persistentDispatcher.addEntryToReplay(entry3);
+ }
+ List<Entry> entries = allEntries.stream()
+ .filter(entry -> positions.contains(entry.getPosition()))
+ .toList();
+ AsyncCallbacks.ReadEntriesCallback callback = invocationOnMock.getArgument(1);
+ Object ctx = invocationOnMock.getArgument(2);
+ callback.readEntriesComplete(copyEntries(entries), ctx);
+ return Collections.emptySet();
+ }).when(cursorMock).asyncReplayEntries(anySet(), any(), any(), anyBoolean());
+
+ CountDownLatch consumer1ReceivedMessage3 = new CountDownLatch(1);
+ mockSendMessages(consumerMock, entries -> {
+ boolean message3Found = entries.stream()
+ .anyMatch(entry -> entry.getPosition().equals(entry3.getPosition()));
+ if (message3Found) {
+ consumer1ReceivedMessage3.countDown();
+ }
+ });
+ mockSendMessages(slowConsumerMock, entries -> { });
+
+ // seed the replay queue with the slow consumer's entries
+ for (Entry entry : List.of(entry1, entry2)) {
+ // add extra retain since addEntryToReplay will release it
+ ((EntryImpl) entry).retain();
+ persistentDispatcher.addEntryToReplay(entry);
+ }
+
+ // trigger the replay read. The batch gets fully discarded since the slow consumer runs out of permits while
+ // the read is in flight. The message for consumer1 that was added to the replay queue in the meantime must
+ // get dispatched by the follow-up read instead of the dispatcher parking a normal read at the end of the
+ // topic.
+ persistentDispatcher.readMoreEntries();
+
+ assertTrue(consumer1ReceivedMessage3.await(5, TimeUnit.SECONDS),
+ "The replayed message for consumer1 with available permits should have been dispatched");
+
+ allEntries.forEach(Entry::release);
+ }
+
@Test(timeOut = 30000)
public void testMessageRedelivery() throws Exception {
final List<Position> actualEntriesToConsumer1 = new CopyOnWriteArrayList<>();
(pulsar) 10/11: [fix][meta] Record get op stats on the correct completion branch in AbstractMetadataStore (#26201)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b3682742836aa2a6cf40ea58af2d87500b71bac8
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Sat Jul 25 19:15:45 2026 +0800
[fix][meta] Record get op stats on the correct completion branch in AbstractMetadataStore (#26201)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit 78cc3f41c510b17d5b2569811e62c5744280401e)
---
.../metadata/impl/AbstractMetadataStore.java | 2 +-
.../impl/AbstractMetadataStoreGetStatsTest.java | 122 +++++++++++++++++++++
2 files changed, 123 insertions(+), 1 deletion(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
index b3e24aa43641..a463e0f1aec9 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
@@ -346,9 +346,9 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
return storeGet(path)
.whenComplete((v, t) -> {
if (t != null) {
- v.ifPresent(getResult -> nodeSizeStats.recordGetRes(path, getResult));
metadataStoreStats.recordGetOpsFailed(System.currentTimeMillis() - start);
} else {
+ v.ifPresent(getResult -> nodeSizeStats.recordGetRes(path, getResult));
metadataStoreStats.recordGetOpsSucceeded(System.currentTimeMillis() - start);
}
});
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/AbstractMetadataStoreGetStatsTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/AbstractMetadataStoreGetStatsTest.java
new file mode 100644
index 000000000000..a90bf4353e72
--- /dev/null
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/AbstractMetadataStoreGetStatsTest.java
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.metadata.impl;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+import io.opentelemetry.api.OpenTelemetry;
+import io.prometheus.client.CollectorRegistry;
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.function.Function;
+import lombok.Cleanup;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataNodeSizeStats;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.metadata.api.Stat;
+import org.apache.pulsar.metadata.api.extended.CreateOption;
+import org.testng.annotations.Test;
+
+/**
+ * Verifies that {@link AbstractMetadataStore#get(String)} records operation stats on the
+ * correct completion branch.
+ */
+public class AbstractMetadataStoreGetStatsTest {
+
+ private static class TestMetadataStore extends AbstractMetadataStore {
+ private volatile Function<String, CompletableFuture<Optional<GetResult>>> getImpl;
+
+ TestMetadataStore(String name, MetadataNodeSizeStats nodeSizeStats) {
+ super(name, OpenTelemetry.noop(), nodeSizeStats, 1);
+ }
+
+ @Override
+ protected CompletableFuture<Optional<GetResult>> storeGet(String path) {
+ return getImpl.apply(path);
+ }
+
+ @Override
+ public CompletableFuture<List<String>> getChildrenFromStore(String path) {
+ return CompletableFuture.completedFuture(List.of());
+ }
+
+ @Override
+ protected CompletableFuture<Boolean> existsFromStore(String path) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ @Override
+ protected CompletableFuture<Void> storeDelete(String path, Optional<Long> expectedVersion) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ protected CompletableFuture<Stat> storePut(String path, byte[] data, Optional<Long> optExpectedVersion,
+ EnumSet<CreateOption> options) {
+ return FutureUtil.failedFuture(new UnsupportedOperationException());
+ }
+ }
+
+ private static double getOpsCount(String storeName, String status) {
+ Double value = CollectorRegistry.defaultRegistry.getSampleValue(
+ "pulsar_metadata_store_ops_latency_ms_count",
+ new String[]{"name", "type", "status"},
+ new String[]{storeName, "get", status});
+ return value == null ? 0.0 : value;
+ }
+
+ @Test
+ public void testGetSuccessRecordsNodeSizeStats() throws Exception {
+ MetadataNodeSizeStats nodeSizeStats = mock(MetadataNodeSizeStats.class);
+ @Cleanup
+ TestMetadataStore store = new TestMetadataStore("get-stats-success", nodeSizeStats);
+ GetResult result = new GetResult(new byte[]{1, 2, 3}, new Stat("/a", 0, 0, 0, false, false));
+ store.getImpl = path -> CompletableFuture.completedFuture(Optional.of(result));
+
+ assertSame(store.get("/a").join().orElse(null), result);
+
+ verify(nodeSizeStats).recordGetRes("/a", result);
+ assertEquals(getOpsCount("get-stats-success", "success"), 1.0);
+ }
+
+ @Test
+ public void testGetFailureRecordsFailedOpAndPropagatesOriginalException() throws Exception {
+ String storeName = "get-stats-failure";
+ @Cleanup
+ TestMetadataStore store = new TestMetadataStore(storeName, mock(MetadataNodeSizeStats.class));
+ MetadataStoreException injected = new MetadataStoreException("injected failure");
+ store.getImpl = path -> FutureUtil.failedFuture(injected);
+
+ CompletionException ex = expectThrows(CompletionException.class, () -> store.get("/a").join());
+
+ assertSame(ex.getCause(), injected);
+ assertTrue(Arrays.stream(injected.getSuppressed()).noneMatch(s -> s instanceof NullPointerException),
+ "stats callback must not throw NPE on the failure path");
+ assertEquals(getOpsCount(storeName, "fail"), 1.0);
+ }
+}
(pulsar) 08/11: [fix][meta] Fix RocksdbMetadataStore instanceId not advancing across restarts (#26218)
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.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d195ab3f948e4b0f0187df75e1d198e8cba036d7
Author: SongOf <46...@users.noreply.github.com>
AuthorDate: Sat Jul 25 16:38:35 2026 +0800
[fix][meta] Fix RocksdbMetadataStore instanceId not advancing across restarts (#26218)
Co-authored-by: maxlisongsong <ma...@didiglobal.com>
(cherry picked from commit a4b7aaa616959af35e6bcd381a24fdf076acd3ac)
---
.../main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
index 08e5478ffcca..35b914e665b0 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/RocksdbMetadataStore.java
@@ -264,7 +264,7 @@ public class RocksdbMetadataStore extends AbstractMetadataStore {
if (value != null) {
generator.set(toLong(value));
} else {
- db.put(writeOptions, INSTANCE_ID_KEY, toBytes(generator.get()));
+ db.put(writeOptions, SEQUENTIAL_ID_KEY, toBytes(generator.get()));
}
return generator;
}