You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2026/05/13 02:51:14 UTC
(pulsar) branch branch-4.0 updated (d86cda88bc7 -> 1cbe883b80b)
This is an automated email from the ASF dual-hosted git repository.
coderzc pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from d86cda88bc7 [fix][broker] Merge broker offload extra configurations (#25736)
new 6d7a22bfc57 [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage (#25514)
new 22637effb89 [fix][broker] Decrement unacked counter when removeAllUpTo removes pending acks (#25581)
new 360a2163a34 [fix][test] Fix flaky testGetExcludedBookiesWithIsolationGroups (#25640)
new 6d4979160a0 [fix][test] Make NamespacesTest.cleanupAfterMethod tolerant of transient infra failures (#25641)
new 1cbe883b80b [fix][test] Fix flaky OneWayReplicatorDeduplicationTest.testDeduplication (#25679)
The 5 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:
.../mledger/impl/LedgerMetadataUtils.java | 30 ++--
.../IsolatedBookieEnsemblePlacementPolicyTest.java | 7 +
.../org/apache/pulsar/broker/service/Consumer.java | 29 ++++
.../pulsar/broker/service/PendingAcksMap.java | 22 ++-
.../PersistentDispatcherMultipleConsumers.java | 4 +-
.../service/schema/BookkeeperSchemaStorage.java | 32 +++-
.../apache/pulsar/broker/admin/NamespacesTest.java | 14 +-
.../service/OneWayReplicatorDeduplicationTest.java | 1 +
.../pulsar/broker/service/PendingAcksMapTest.java | 31 +++-
.../pulsar/broker/stats/ConsumerStatsTest.java | 166 +++++++++++++++++++++
.../pulsar/client/api/ProducerConsumerBase.java | 3 +-
.../java/org/apache/pulsar/schema/SchemaTest.java | 65 ++++++++
.../bookkeeper/client/PulsarMockBookKeeper.java | 2 +-
.../bookkeeper/client/PulsarMockLedgerHandle.java | 22 ++-
14 files changed, 386 insertions(+), 42 deletions(-)
(pulsar) 04/05: [fix][test] Make NamespacesTest.cleanupAfterMethod tolerant of transient infra failures (#25641)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
coderzc pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6d4979160a051dac02b733e55dd076decbb90513
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri May 1 06:11:51 2026 -0700
[fix][test] Make NamespacesTest.cleanupAfterMethod tolerant of transient infra failures (#25641)
---
.../org/apache/pulsar/broker/admin/NamespacesTest.java | 14 ++++++++++----
1 file changed, 10 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
index da3ced47efc..5048fc4ca08 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -187,12 +187,18 @@ public class NamespacesTest extends MockedPulsarServiceBaseTest {
}
@AfterMethod(alwaysRun = true)
- public void cleanupAfterMethod() throws Exception{
- // cleanup.
+ public void cleanupAfterMethod() {
+ // Best-effort cleanup. Transient infra failures (e.g. ZK session expiry from a
+ // long-running test or GC pause) can make the admin call fail here; log and
+ // continue so the test method's actual result is preserved.
Set<String> existsNsSetAferSetup = Stream.concat(testLocalNamespaces.stream(), testGlobalNamespaces.stream())
.map(Objects::toString).collect(Collectors.toSet());
- cleanupNamespaceByPredicate(this.testTenant, v -> !existsNsSetAferSetup.contains(v));
- cleanupNamespaceByPredicate(this.testOtherTenant, v -> !existsNsSetAferSetup.contains(v));
+ try {
+ cleanupNamespaceByPredicate(this.testTenant, v -> !existsNsSetAferSetup.contains(v));
+ cleanupNamespaceByPredicate(this.testOtherTenant, v -> !existsNsSetAferSetup.contains(v));
+ } catch (Exception e) {
+ log.warn().exception(e).log("Failed to clean up namespaces after test method");
+ }
}
protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) {
(pulsar) 02/05: [fix][broker] Decrement unacked counter when removeAllUpTo removes pending acks (#25581)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
coderzc pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 22637effb895aec8722f5b30227519671c7b1d5a
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Mon Apr 27 20:50:37 2026 +0800
[fix][broker] Decrement unacked counter when removeAllUpTo removes pending acks (#25581)
---
.../org/apache/pulsar/broker/service/Consumer.java | 29 ++++
.../pulsar/broker/service/PendingAcksMap.java | 22 ++-
.../PersistentDispatcherMultipleConsumers.java | 4 +-
.../pulsar/broker/service/PendingAcksMapTest.java | 31 +++-
.../pulsar/broker/stats/ConsumerStatsTest.java | 166 +++++++++++++++++++++
.../pulsar/client/api/ProducerConsumerBase.java | 3 +-
6 files changed, 243 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
index 1038baf6ba2..c0b27b22382 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -1102,6 +1102,35 @@ public class Consumer {
return pendingAcks;
}
+ /**
+ * Remove all pending acks up to the given mark-delete position and decrement the consumer's unacked message
+ * counter by the remaining unacked count for each removed entry.
+ *
+ * <p>This is used when the cursor's mark-delete position advances past entries that are still in the consumer's
+ * pending acks. The remaining unacked count accounts for batch index level acknowledgments — only the truly
+ * unacked batch indexes are decremented.
+ *
+ * @param markDeleteLedgerId the ledger ID up to which to remove pending acks
+ * @param markDeleteEntryId the entry ID up to which to remove pending acks
+ */
+ public void removePendingAcksUpToPositionAndDecrementUnacked(long markDeleteLedgerId, long markDeleteEntryId) {
+ if (pendingAcks == null) {
+ return;
+ }
+
+ MutableInt mutableTotalUnacked = new MutableInt(0);
+ pendingAcks.removeAllUpTo(markDeleteLedgerId, markDeleteEntryId,
+ (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ mutableTotalUnacked.add((int) getUnAckedCountForBatchIndexLevelEnabled(
+ PositionFactory.create(ledgerId, entryId), batchSize));
+ });
+ int totalUnacked = mutableTotalUnacked.intValue();
+ if (totalUnacked > 0) {
+ addAndGetUnAckedMsgs(this, -totalUnacked);
+ updateBlockedConsumerOnUnackedMsgs(this);
+ }
+ }
+
public int getPriorityLevel() {
return priorityLevel;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java
index 7a728a037dc..0f7802d8294 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java
@@ -326,13 +326,16 @@ public class PendingAcksMap {
}
/**
- * Remove all pending acks up to the given ledger ID and entry ID.
+ * Remove all pending acks up to the given ledger ID and entry ID, invoking a callback for each removed entry.
*
* @param markDeleteLedgerId the ledger ID up to which to remove pending acks
* @param markDeleteEntryId the entry ID up to which to remove pending acks
+ * @param removedEntryCallback optional callback invoked for each removed entry (within the write lock),
+ * receiving ledgerId, entryId, batchSize, and stickyKeyHash
*/
- public void removeAllUpTo(long markDeleteLedgerId, long markDeleteEntryId) {
- internalRemoveAllUpTo(markDeleteLedgerId, markDeleteEntryId, false);
+ public void removeAllUpTo(long markDeleteLedgerId, long markDeleteEntryId,
+ PendingAcksConsumer removedEntryCallback) {
+ internalRemoveAllUpTo(markDeleteLedgerId, markDeleteEntryId, false, removedEntryCallback);
}
/**
@@ -345,8 +348,10 @@ public class PendingAcksMap {
* @param markDeleteLedgerId the ledger ID up to which to remove pending acks
* @param markDeleteEntryId the entry ID up to which to remove pending acks
* @param useWriteLock true if the method should use a write lock, false otherwise
+ * @param removedEntryCallback optional callback invoked for each removed entry (within the write lock)
*/
- private void internalRemoveAllUpTo(long markDeleteLedgerId, long markDeleteEntryId, boolean useWriteLock) {
+ private void internalRemoveAllUpTo(long markDeleteLedgerId, long markDeleteEntryId, boolean useWriteLock,
+ PendingAcksConsumer removedEntryCallback) {
PendingAcksRemoveHandler pendingAcksRemoveHandler = pendingAcksRemoveHandlerSupplier.get();
// track if the write lock was acquired
boolean acquiredWriteLock = false;
@@ -382,14 +387,19 @@ public class PendingAcksMap {
retryWithWriteLock = true;
return;
}
+ IntIntPair value = intIntPairEntry.getValue();
+ int batchSize = value.leftInt();
+ int stickyKeyHash = value.rightInt();
if (pendingAcksRemoveHandler != null) {
if (!batchStarted) {
pendingAcksRemoveHandler.startBatch();
batchStarted = true;
}
- int stickyKeyHash = intIntPairEntry.getValue().rightInt();
pendingAcksRemoveHandler.handleRemoving(consumer, ledgerId, entryId, stickyKeyHash, closed);
}
+ if (removedEntryCallback != null) {
+ removedEntryCallback.accept(ledgerId, entryId, batchSize, stickyKeyHash);
+ }
entryMapIterator.remove();
}
if (ledgerMap.isEmpty()) {
@@ -409,7 +419,7 @@ public class PendingAcksMap {
} else {
readLock.unlock();
if (retryWithWriteLock) {
- internalRemoveAllUpTo(markDeleteLedgerId, markDeleteEntryId, true);
+ internalRemoveAllUpTo(markDeleteLedgerId, markDeleteEntryId, true, removedEntryCallback);
}
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index 7d578fbe3be..59fd48a8079 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -363,8 +363,8 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
if (lastMarkDeletePositionBeforeReadMoreEntries != markDeletePosition) {
redeliveryMessages.removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
for (Consumer consumer : consumerList) {
- consumer.getPendingAcks()
- .removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
+ consumer.removePendingAcksUpToPositionAndDecrementUnacked(
+ markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
}
lastMarkDeletePositionBeforeReadMoreEntries = markDeletePosition;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java
index 42f5935ca88..8db0e3a0f73 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PendingAcksMapTest.java
@@ -117,7 +117,8 @@ public class PendingAcksMapTest {
pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
pendingAcksMap.addPendingAckIfAllowed(2L, 1L, 1, 125);
- pendingAcksMap.removeAllUpTo(1L, 2L);
+ pendingAcksMap.removeAllUpTo(1L, 2L, (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ });
assertFalse(pendingAcksMap.contains(1L, 1L));
assertFalse(pendingAcksMap.contains(1L, 2L));
@@ -134,7 +135,8 @@ public class PendingAcksMapTest {
pendingAcksMap.addPendingAckIfAllowed(2L, 2L, 1, 126);
pendingAcksMap.addPendingAckIfAllowed(3L, 1L, 1, 127);
- pendingAcksMap.removeAllUpTo(2L, 1L);
+ pendingAcksMap.removeAllUpTo(2L, 1L, (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ });
assertFalse(pendingAcksMap.contains(1L, 1L));
assertFalse(pendingAcksMap.contains(1L, 2L));
@@ -176,13 +178,36 @@ public class PendingAcksMapTest {
pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 1, 124);
pendingAcksMap.addPendingAckIfAllowed(2L, 1L, 1, 125);
- pendingAcksMap.removeAllUpTo(1L, 2L);
+ pendingAcksMap.removeAllUpTo(1L, 2L, (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ });
verify(removeHandler).handleRemoving(consumer, 1L, 1L, 123, false);
verify(removeHandler).handleRemoving(consumer, 1L, 2L, 124, false);
verify(removeHandler, never()).handleRemoving(consumer, 2L, 1L, 125, false);
}
+ @Test
+ public void removeAllUpToWithCallback_InvokesCallbackForEachRemovedEntry() {
+ Consumer consumer = createMockConsumer("consumer1");
+ PendingAcksMap pendingAcksMap = new PendingAcksMap(consumer, () -> null, () -> null);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 1L, 3, 123);
+ pendingAcksMap.addPendingAckIfAllowed(1L, 2L, 5, 124);
+ pendingAcksMap.addPendingAckIfAllowed(2L, 1L, 7, 125);
+
+ List<int[]> callbackInvocations = new ArrayList<>();
+ pendingAcksMap.removeAllUpTo(1L, 2L,
+ (ledgerId, entryId, batchSize, stickyKeyHash) -> {
+ callbackInvocations.add(new int[]{(int) ledgerId, (int) entryId, batchSize, stickyKeyHash});
+ });
+
+ assertEquals(callbackInvocations.size(), 2);
+ assertEquals(callbackInvocations.get(0), new int[]{1, 1, 3, 123});
+ assertEquals(callbackInvocations.get(1), new int[]{1, 2, 5, 124});
+ assertFalse(pendingAcksMap.contains(1L, 1L));
+ assertFalse(pendingAcksMap.contains(1L, 2L));
+ assertTrue(pendingAcksMap.contains(2L, 1L));
+ }
+
@Test
public void size_ReturnsCorrectSize() {
Consumer consumer = createMockConsumer("consumer1");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
index ae2da5b6598..6e6b6d31f6e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ConsumerStatsTest.java
@@ -59,6 +59,7 @@ import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
import org.apache.pulsar.broker.service.StickyKeyDispatcher;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.plugin.EntryFilter;
import org.apache.pulsar.broker.service.plugin.EntryFilterProducerTest;
@@ -106,6 +107,7 @@ public class ConsumerStatsTest extends ProducerConsumerBase {
@Override
protected ServiceConfiguration getDefaultConf() {
ServiceConfiguration conf = super.getDefaultConf();
+ conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
conf.setMaxUnackedMessagesPerConsumer(0);
// wait for shutdown of the broker, this prevents flakiness which could be caused by metrics being
// unregistered asynchronously. This impacts the execution of the next test method if this would be happening.
@@ -727,6 +729,170 @@ public class ConsumerStatsTest extends ProducerConsumerBase {
}
+ @DataProvider(name = "subscriptionTypes")
+ public Object[][] subscriptionTypes() {
+ return new Object[][]{
+ {SubscriptionType.Shared},
+ {SubscriptionType.Key_Shared}
+ };
+ }
+
+ /**
+ * Verify unacked count is correctly decremented when removeAllUpTo removes non-batch
+ * entries from pendingAcks after mark-delete advances via message expiry.
+ */
+ @Test(dataProvider = "subscriptionTypes")
+ public void testUnackedCountNonBatchAfterExpire(SubscriptionType subType) throws Exception {
+ String topic = newTopicName();
+ String sub = "sub";
+ int numMessages = 10;
+
+ @Cleanup Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic).enableBatching(false).create();
+ @Cleanup Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic).subscriptionName(sub)
+ .subscriptionType(subType)
+ .subscribe();
+
+ for (int i = 0; i < numMessages; i++) {
+ producer.send(("msg-" + i).getBytes());
+ }
+
+ org.apache.pulsar.broker.service.Consumer svcConsumer =
+ getTheUniqueServiceConsumer(topic, sub);
+ for (int i = 0; i < numMessages; i++) {
+ Message<byte[]> msg = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(msg, "Expected to receive message " + i);
+ }
+
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(numMessages, svcConsumer.getUnackedMessages()));
+
+ expireAndVerifyUnackedDrained(topic, sub, producer, consumer, svcConsumer);
+ }
+
+ /**
+ * Verify unacked count is correctly decremented when removeAllUpTo removes batch
+ * entries from pendingAcks after mark-delete advances via message expiry.
+ */
+ @Test(dataProvider = "subscriptionTypes")
+ public void testUnackedCountBatchAfterExpire(SubscriptionType subType) throws Exception {
+ String topic = newTopicName();
+ String sub = "sub";
+ int numMessages = 10;
+
+ @Cleanup Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .batchingMaxMessages(20)
+ .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+ .enableBatching(true)
+ .create();
+ @Cleanup Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic).subscriptionName(sub)
+ .subscriptionType(subType)
+ .subscribe();
+
+ for (int i = 0; i < numMessages; i++) {
+ producer.newMessage().value(("batch-" + i).getBytes()).sendAsync();
+ }
+ producer.flush();
+
+ for (int i = 0; i < numMessages; i++) {
+ Message<byte[]> msg = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(msg, "Expected to receive message " + i);
+ }
+
+ org.apache.pulsar.broker.service.Consumer svcConsumer =
+ getTheUniqueServiceConsumer(topic, sub);
+
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(numMessages, svcConsumer.getUnackedMessages()));
+
+ expireAndVerifyUnackedDrained(topic, sub, producer, consumer, svcConsumer);
+ }
+
+ /**
+ * Verify unacked count is correctly decremented when removeAllUpTo removes a partially-acked
+ * batch entry from pendingAcks after mark-delete advances via message expiry.
+ *
+ * <p>Flow: produce batch(batchSize=10) → consume all → ack 5 of 10 → expire → unacked should be 0.
+ */
+ @Test(dataProvider = "subscriptionTypes")
+ public void testUnackedCountBatchPartialAckAfterExpire(SubscriptionType subType) throws Exception {
+ String topic = newTopicName();
+ String sub = "sub";
+ int numMessages = 10;
+ int ackCount = 5;
+
+ @Cleanup Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topic)
+ .batchingMaxMessages(20)
+ .batchingMaxPublishDelay(1, TimeUnit.HOURS)
+ .enableBatching(true)
+ .create();
+ @Cleanup Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(sub)
+ .enableBatchIndexAcknowledgment(true)
+ .subscriptionType(subType)
+ .subscribe();
+
+ for (int i = 0; i < numMessages; i++) {
+ producer.newMessage().value(("batch-" + i).getBytes()).sendAsync();
+ }
+ producer.flush();
+
+ List<Message<byte[]>> messages = new ArrayList<>();
+ for (int i = 0; i < numMessages; i++) {
+ Message<byte[]> msg = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(msg, "Expected to receive message " + i);
+ messages.add(msg);
+ }
+
+ org.apache.pulsar.broker.service.Consumer svcConsumer =
+ getTheUniqueServiceConsumer(topic, sub);
+
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(numMessages, svcConsumer.getUnackedMessages()));
+
+ // Partially ack — ack 5 of 10 batch indexes
+ for (int i = 0; i < ackCount; i++) {
+ consumer.acknowledge(messages.get(i));
+ }
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(numMessages - ackCount, svcConsumer.getUnackedMessages()));
+
+ expireAndVerifyUnackedDrained(topic, sub, producer, consumer, svcConsumer);
+ }
+
+ private void expireAndVerifyUnackedDrained(String topic, String sub,
+ Producer<byte[]> producer, Consumer<byte[]> consumer,
+ org.apache.pulsar.broker.service.Consumer svcConsumer)
+ throws Exception {
+ PersistentTopic pTopic = (PersistentTopic) pulsar.getBrokerService()
+ .getTopicReference(topic).get();
+
+ Thread.sleep(1100);
+ pTopic.getSubscription(sub).expireMessagesAsync(1).get();
+
+ // Trigger readMoreEntries to invoke removeAllUpTo
+ producer.send("trigger".getBytes());
+ Message<byte[]> triggerMsg = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(triggerMsg);
+ consumer.acknowledge(triggerMsg);
+
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
+ assertEquals(0, svcConsumer.getUnackedMessages()));
+ }
+
+ private org.apache.pulsar.broker.service.Consumer getTheUniqueServiceConsumer(String topic, String sub) {
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(topic, false).join().get();
+ AbstractPersistentDispatcherMultipleConsumers dispatcher =
+ (AbstractPersistentDispatcherMultipleConsumers) persistentTopic.getSubscription(sub).getDispatcher();
+ return dispatcher.getConsumers().iterator().next();
+ }
+
private String findConsumerNameForHash(SubscriptionStats subscriptionStats, int hash) {
return findConsumerForHash(subscriptionStats, hash).map(ConsumerStats::getConsumerName).orElse(null);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java
index 01d2f107dff..1a963b523c2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java
@@ -30,6 +30,7 @@ import java.util.Set;
import java.util.function.BiFunction;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.testng.Assert;
@@ -71,7 +72,7 @@ public abstract class ProducerConsumerBase extends MockedPulsarServiceBaseTest {
private static final Random random = new Random();
protected String newTopicName() {
- return "my-property/my-ns/topic-" + Long.toHexString(random.nextLong());
+ return TopicName.get("my-property/my-ns/topic-" + Long.toHexString(random.nextLong())).toString();
}
protected <T> ReceivedMessages<T> receiveAndAckMessages(
(pulsar) 03/05: [fix][test] Fix flaky testGetExcludedBookiesWithIsolationGroups (#25640)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
coderzc pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 360a2163a34b1e7b6319f6d50f87bb8a095b2ec5
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri May 1 06:09:54 2026 -0700
[fix][test] Fix flaky testGetExcludedBookiesWithIsolationGroups (#25640)
---
.../rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java
index 936b04386ff..0dc996c7d7d 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java
@@ -751,6 +751,13 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
+ // Wait for the async cache load triggered by initialize() to complete; otherwise
+ // getExcludedBookiesWithIsolationGroups returns an empty set when cachedRackConfiguration
+ // is still null. Same pattern used in testBookieInfoChange (#25473).
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertNotNull(isolationPolicy.getBookieMappingCache()
+ .getIfCached(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH)));
+
/* Test common cases */
MutablePair<Set<String>, Set<String>> groups = new MutablePair<>();
groups.setLeft(Sets.newHashSet(isolationGroup1));
(pulsar) 01/05: [fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage (#25514)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
coderzc pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6d7a22bfc57bfc1d5f61463320dbb2866b132271
Author: zhou zhuohan <84...@qq.com>
AuthorDate: Thu Apr 23 20:50:39 2026 +0800
[fix][broker] Clean up orphan ledger on concurrent initial schema creation in BookkeeperSchemaStorage (#25514)
---
.../mledger/impl/LedgerMetadataUtils.java | 30 +++++-----
.../service/schema/BookkeeperSchemaStorage.java | 32 +++++++++--
.../java/org/apache/pulsar/schema/SchemaTest.java | 65 ++++++++++++++++++++++
.../bookkeeper/client/PulsarMockBookKeeper.java | 2 +-
.../bookkeeper/client/PulsarMockLedgerHandle.java | 22 ++++++--
5 files changed, 125 insertions(+), 26 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/LedgerMetadataUtils.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/LedgerMetadataUtils.java
index 4ac409a2e9b..13157ad4e38 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/LedgerMetadataUtils.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/LedgerMetadataUtils.java
@@ -29,28 +29,28 @@ import org.apache.pulsar.common.policies.data.EnsemblePlacementPolicyConfig;
*/
public final class LedgerMetadataUtils {
- private static final String METADATA_PROPERTY_APPLICATION = "application";
- private static final byte[] METADATA_PROPERTY_APPLICATION_PULSAR = "pulsar".getBytes(StandardCharsets.UTF_8);
+ public static final String METADATA_PROPERTY_APPLICATION = "application";
+ public static final byte[] METADATA_PROPERTY_APPLICATION_PULSAR = "pulsar".getBytes(StandardCharsets.UTF_8);
- private static final String METADATA_PROPERTY_COMPONENT = "component";
- private static final byte[] METADATA_PROPERTY_COMPONENT_MANAGED_LEDGER =
+ public static final String METADATA_PROPERTY_COMPONENT = "component";
+ public static final byte[] METADATA_PROPERTY_COMPONENT_MANAGED_LEDGER =
"managed-ledger".getBytes(StandardCharsets.UTF_8);
- private static final byte[] METADATA_PROPERTY_COMPONENT_COMPACTED_LEDGER =
+ public static final byte[] METADATA_PROPERTY_COMPONENT_COMPACTED_LEDGER =
"compacted-ledger".getBytes(StandardCharsets.UTF_8);
- private static final byte[] METADATA_PROPERTY_COMPONENT_SCHEMA = "schema".getBytes(StandardCharsets.UTF_8);
+ public static final byte[] METADATA_PROPERTY_COMPONENT_SCHEMA = "schema".getBytes(StandardCharsets.UTF_8);
- private static final byte[] METADATA_PROPERTY_COMPONENT_DELAYED_INDEX_BUCKET =
+ public static final byte[] METADATA_PROPERTY_COMPONENT_DELAYED_INDEX_BUCKET =
"delayed-index-bucket".getBytes(StandardCharsets.UTF_8);
- private static final String METADATA_PROPERTY_MANAGED_LEDGER_NAME = "pulsar/managed-ledger";
- private static final String METADATA_PROPERTY_CURSOR_NAME = "pulsar/cursor";
- private static final String METADATA_PROPERTY_COMPACTEDTOPIC = "pulsar/compactedTopic";
- private static final String METADATA_PROPERTY_COMPACTEDTO = "pulsar/compactedTo";
- private static final String METADATA_PROPERTY_SCHEMAID = "pulsar/schemaId";
+ public static final String METADATA_PROPERTY_MANAGED_LEDGER_NAME = "pulsar/managed-ledger";
+ public static final String METADATA_PROPERTY_CURSOR_NAME = "pulsar/cursor";
+ public static final String METADATA_PROPERTY_COMPACTEDTOPIC = "pulsar/compactedTopic";
+ public static final String METADATA_PROPERTY_COMPACTEDTO = "pulsar/compactedTo";
+ public static final String METADATA_PROPERTY_SCHEMAID = "pulsar/schemaId";
- private static final String METADATA_PROPERTY_DELAYED_INDEX_BUCKET_KEY = "pulsar/delayedIndexBucketKey";
- private static final String METADATA_PROPERTY_DELAYED_INDEX_TOPIC = "pulsar/delayedIndexTopic";
- private static final String METADATA_PROPERTY_DELAYED_INDEX_CURSOR = "pulsar/delayedIndexCursor";
+ public static final String METADATA_PROPERTY_DELAYED_INDEX_BUCKET_KEY = "pulsar/delayedIndexBucketKey";
+ public static final String METADATA_PROPERTY_DELAYED_INDEX_TOPIC = "pulsar/delayedIndexTopic";
+ public static final String METADATA_PROPERTY_DELAYED_INDEX_CURSOR = "pulsar/delayedIndexCursor";
/**
* Build base metadata for every ManagedLedger.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
index e38bf48f1fd..ea6537e1e1e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -351,7 +351,10 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
.setLedgerId(-1L)
).build();
- return addNewSchemaEntryToStore(schemaId, Collections.singletonList(emptyIndex), data).thenCompose(position -> {
+ CompletableFuture<SchemaStorageFormat.PositionInfo> stored =
+ addNewSchemaEntryToStore(schemaId, Collections.singletonList(emptyIndex), data);
+
+ return stored.thenCompose(position -> {
// The schema was stored in the ledger, now update the z-node with the pointer to it
SchemaStorageFormat.IndexEntry info = SchemaStorageFormat.IndexEntry.newBuilder()
.setVersion(0)
@@ -359,12 +362,30 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
.setHash(copyFrom(hash))
.build();
- return createSchemaLocator(getSchemaPath(schemaId), SchemaStorageFormat.SchemaLocator.newBuilder()
+ CompletableFuture<Long> created = createSchemaLocator(getSchemaPath(schemaId),
+ SchemaStorageFormat.SchemaLocator.newBuilder()
.setInfo(info)
.addAllIndex(
newArrayList(info))
.build())
- .thenApply(ignore -> 0L);
+ .thenApply(ignore -> 0L);
+
+ return created.whenComplete((__, ex) -> {
+ if (ex == null) {
+ return;
+ }
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.warn("[{}] Failed to create schema locator with ledger {}", schemaId, position.getLedgerId(),
+ cause);
+ if (cause instanceof AlreadyExistsException || cause instanceof BadVersionException) {
+ bookKeeper.asyncDeleteLedger(position.getLedgerId(), (rc, ctx) -> {
+ if (rc != BKException.Code.OK) {
+ log.warn("[{}] Failed to delete orphan ledger {} after schema locator creation failed, "
+ + "rc: {}", schemaId, position.getLedgerId(), rc);
+ }
+ }, null);
+ }
+ });
});
}
@@ -478,7 +499,8 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
).thenApply(ignore -> nextVersion).whenComplete((__, ex) -> {
if (ex != null) {
Throwable cause = FutureUtil.unwrapCompletionException(ex);
- log.warn("[{}] Failed to update schema locator with position {}", schemaId, position, cause);
+ log.warn("[{}] Failed to update schema locator with ledger {}", schemaId, position.getLedgerId(),
+ cause);
if (cause instanceof AlreadyExistsException || cause instanceof BadVersionException) {
bookKeeper.asyncDeleteLedger(position.getLedgerId(), new AsyncCallback.DeleteCallback() {
@Override
@@ -762,4 +784,4 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
throw t instanceof CompletionException ? (CompletionException) t : new CompletionException(t);
});
}
-}
\ No newline at end of file
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
index 9bf1f5e4048..286937cc42f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
@@ -57,6 +57,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema.Parser;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.mledger.impl.LedgerMetadataUtils;
import org.apache.pulsar.broker.BrokerTestUtil;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage;
@@ -1471,6 +1472,70 @@ public class SchemaTest extends MockedPulsarServiceBaseTest {
producers2.clear();
}
+ /**
+ * Test that when multiple producers concurrently create schema for a brand-new topic,
+ * the orphan BookKeeper ledgers created by the losing requests are properly cleaned up.
+ */
+ @Test
+ public void testConcurrentCreateSchemaNoOrphanLedger() throws Exception {
+ final String namespace = "test-namespace-" + randomName(16);
+ String ns = PUBLIC_TENANT + "/" + namespace;
+ admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
+
+ final String topic = getTopicName(ns, "testConcurrentCreateSchemaNoOrphanLedger");
+ final String schemaName = TopicName.get(topic).getSchemaName();
+
+ org.apache.bookkeeper.client.PulsarMockBookKeeper mockBk =
+ (org.apache.bookkeeper.client.PulsarMockBookKeeper) pulsar.getBookKeeperClient();
+
+ // Concurrently create producers with the same schema on a brand-new topic
+ int concurrency = 16;
+ @Cleanup("shutdownNow")
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<CompletableFuture<Producer<Schemas.PersonOne>>> producers =
+ Collections.synchronizedList(new ArrayList<>(concurrency));
+ CountDownLatch latch = new CountDownLatch(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ executor.execute(() -> {
+ try {
+ producers.add(pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic).createAsync());
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+ latch.await();
+ FutureUtil.waitForAll(producers).join();
+
+ // Verify only 1 schema version exists
+ assertEquals(admin.schemas().getAllSchemas(topic).size(), 1);
+
+ // Count surviving BK ledgers whose customMetadata "pulsar/schemaId" matches this topic's schemaName.
+ // If orphan ledgers were not cleaned up, there would be more than 1.
+ int schemaLedgerCount = 0;
+ for (org.apache.bookkeeper.client.PulsarMockLedgerHandle lh : mockBk.getLedgerMap().values()) {
+ Map<String, byte[]> metadata = lh.getLedgerMetadata().getCustomMetadata();
+ byte[] schemaIdBytes = metadata.get(LedgerMetadataUtils.METADATA_PROPERTY_SCHEMAID);
+ if (schemaIdBytes != null
+ && schemaName.equals(new String(schemaIdBytes, java.nio.charset.StandardCharsets.UTF_8))) {
+ schemaLedgerCount++;
+ }
+ }
+ assertEquals(schemaLedgerCount, 1,
+ "Expected exactly 1 schema ledger for the topic, but found "
+ + schemaLedgerCount + ". Orphan ledgers were not cleaned up.");
+
+ // Cleanup
+ producers.forEach(p -> {
+ try {
+ p.join().close();
+ } catch (Exception ignore) {
+ }
+ });
+ producers.clear();
+ }
+
@EqualsAndHashCode
static class User implements Serializable {
private String name;
diff --git a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
index 54b8853fb0d..7104ded7460 100644
--- a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
+++ b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockBookKeeper.java
@@ -133,7 +133,7 @@ public class PulsarMockBookKeeper extends BookKeeper {
long id = sequence.getAndIncrement();
log.info("Creating ledger {}", id);
PulsarMockLedgerHandle lh =
- new PulsarMockLedgerHandle(PulsarMockBookKeeper.this, id, digestType, passwd);
+ new PulsarMockLedgerHandle(PulsarMockBookKeeper.this, id, digestType, passwd, properties);
ledgers.put(id, lh);
return FutureUtils.value(lh);
} catch (Throwable t) {
diff --git a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
index 2f5ded0fe01..4d1fd1380c8 100644
--- a/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
+++ b/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
@@ -28,6 +28,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
@@ -69,7 +70,14 @@ public class PulsarMockLedgerHandle extends LedgerHandle {
public PulsarMockLedgerHandle(PulsarMockBookKeeper bk, long id,
DigestType digest, byte[] passwd) throws GeneralSecurityException {
- super(bk.getClientCtx(), id, new Versioned<>(createMetadata(id, digest, passwd), new LongVersion(0L)),
+ this(bk, id, digest, passwd, Collections.emptyMap());
+ }
+
+ public PulsarMockLedgerHandle(PulsarMockBookKeeper bk, long id,
+ DigestType digest, byte[] passwd,
+ Map<String, byte[]> customMetadata) throws GeneralSecurityException {
+ super(bk.getClientCtx(), id,
+ new Versioned<>(createMetadata(id, digest, passwd, customMetadata), new LongVersion(0L)),
digest, passwd, WriteFlag.NONE);
this.bk = bk;
this.id = id;
@@ -268,14 +276,18 @@ public class PulsarMockLedgerHandle extends LedgerHandle {
return readHandle.readLastAddConfirmedAndEntryAsync(entryId, timeOutInMillis, parallel);
}
- private static LedgerMetadata createMetadata(long id, DigestType digest, byte[] passwd) {
+ private static LedgerMetadata createMetadata(long id, DigestType digest, byte[] passwd,
+ Map<String, byte[]> customMetadata) {
List<BookieId> ensemble = new ArrayList<>(PulsarMockBookKeeper.getMockEnsemble());
- return LedgerMetadataBuilder.create()
+ LedgerMetadataBuilder builder = LedgerMetadataBuilder.create()
.withDigestType(digest.toApiDigestType())
.withPassword(passwd)
.withId(id)
- .newEnsembleEntry(0L, ensemble)
- .build();
+ .newEnsembleEntry(0L, ensemble);
+ if (customMetadata != null && !customMetadata.isEmpty()) {
+ builder.withCustomMetadata(customMetadata);
+ }
+ return builder.build();
}
private static final Logger log = LoggerFactory.getLogger(PulsarMockLedgerHandle.class);
(pulsar) 05/05: [fix][test] Fix flaky OneWayReplicatorDeduplicationTest.testDeduplication (#25679)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
coderzc pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1cbe883b80b19667881f96fdd572ff4fa8d20fd4
Author: Praveen Kumar <12...@users.noreply.github.com>
AuthorDate: Wed May 6 17:04:05 2026 +0530
[fix][test] Fix flaky OneWayReplicatorDeduplicationTest.testDeduplication (#25679)
---
.../apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java
index 353d2ab89e2..344279d2537 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/OneWayReplicatorDeduplicationTest.java
@@ -114,6 +114,7 @@ public class OneWayReplicatorDeduplicationTest extends OneWayReplicatorTestBase
LocalBookkeeperEnsemble bookkeeperEnsemble, ZookeeperServerTest brokerConfigZk) {
super.setConfigDefaults(config, clusterName, bookkeeperEnsemble, brokerConfigZk);
// For check whether deduplication snapshot has done.
+ config.setBrokerDeduplicationSnapshotIntervalSeconds(1);
config.setBrokerDeduplicationEntriesInterval(10);
config.setReplicationStartAt("earliest");
// To cover more cases, write more than one ledger.