Skip to content

[fix][broker] Close pending acks cleanup gap in BacklogQuotaManager #25624

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nodece merged 1 commit into from
Apr 30, 2026
Merged

[fix][broker] Close pending acks cleanup gap in BacklogQuotaManager #25624

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 62 additions & 12 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.pulsar.broker.service;

import static java.util.concurrent.TimeUnit.SECONDS;
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
Expand All @@ -33,12 +34,14 @@
import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.resources.NamespaceResources;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.persistent.PersistentTopicMetrics.BacklogQuotaMetrics;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType;
import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.MetadataStoreException;

Expand Down Expand Up @@ -135,9 +138,7 @@ public void handleExceededBacklogQuota(PersistentTopic persistentTopic, BacklogQ
* Backlog quota set for the topic
*/
private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuota quota) {
// Set the reduction factor to 90%. The aim is to drop down the backlog to 90% of the quota limit.
double reductionFactor = 0.9;
double targetSize = reductionFactor * quota.getLimitSize();
long targetSize = computeEvictionTarget(quota.getLimitSize());

// Get estimated unconsumed size for the managed ledger associated with this topic. Estimated size is more
// useful than the actual storage size. Actual storage size gets updated only when managed ledger is trimmed.
Expand All @@ -147,7 +148,7 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
log.debug()
.attr("topic", persistentTopic.getName())
.attr("targetSize", targetSize)
.attr("quotaLimit", targetSize / reductionFactor)
.attr("quotaLimit", quota.getLimitSize())
.attr("backlogSize", backlogSize)
.log("Target size for quota limit");
ManagedCursor previousSlowestConsumer = null;
Expand All @@ -159,21 +160,25 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
log.debug().attr("topic", persistentTopic.getName()).log("Slowest consumer is null");
break;
}
double messageSkipFactor = ((backlogSize - targetSize) / backlogSize);

if (slowestConsumer == previousSlowestConsumer) {
log.info()
.attr("topic", persistentTopic.getName())
.attr("targetSize", targetSize)
.attr("quotaLimit", targetSize / reductionFactor)
.attr("quotaLimit", quota.getLimitSize())
.attr("backlogSize", backlogSize)
.log("Cursors not progressing");
break;
}

// Calculate number of messages to be skipped using the current backlog and the skip factor.
long entriesInBacklog = slowestConsumer.getNumberOfEntriesInBacklog(false);
int messagesToSkip = (int) (messageSkipFactor * entriesInBacklog);

int messagesToSkip = computeEntriesToEvict(
backlogSize,
quota.getLimitSize(),
entriesInBacklog);

try {
// If there are no messages to skip, break out of the loop
if (messagesToSkip == 0) {
Expand All @@ -188,6 +193,7 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
.attr("entriesInBacklog", entriesInBacklog)
.log("Skipping messages on slowest consumer having backlog entries");
slowestConsumer.skipEntries(messagesToSkip, IndividualDeletedEntries.Include);
markDeletePositionMoveForward(persistentTopic, slowestConsumer);
} catch (Exception e) {
log.error()
.attr("topic", persistentTopic.getName())
Expand All @@ -203,8 +209,7 @@ private void dropBacklogForSizeLimit(PersistentTopic persistentTopic, BacklogQuo
log.debug()
.attr("topic", persistentTopic.getName())
.attr("backlogSize", backlogSize)
.attr("messageSkipFactor", messageSkipFactor)
.log("Updated unconsumed size =. skipFactor");
.log("Updated unconsumed size");
}
}

Expand All @@ -220,9 +225,8 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
boolean preciseTimeBasedBacklogQuotaCheck) {
// If enabled precise time based backlog quota check, will expire message based on the timeBaseQuota
if (preciseTimeBasedBacklogQuotaCheck) {
// Set the reduction factor to 90%. The aim is to drop down the backlog to 90% of the quota limit.
double reductionFactor = 0.9;
int target = (int) (reductionFactor * quota.getLimitTime());
int target = (int) computeEvictionTarget(quota.getLimitTime());

log.debug()
.attr("topic", persistentTopic.getName())
.attr("target", target)
Expand Down Expand Up @@ -253,6 +257,7 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
long ledgerId = mLedger.getLedgersInfo().ceilingKey(oldestPosition.getLedgerId() + 1);
Position nextPosition = PositionFactory.create(ledgerId, -1);
slowestConsumer.markDelete(nextPosition);
markDeletePositionMoveForward(persistentTopic, slowestConsumer);
continue;
}
// Timestamp only > 0 if ledger has been closed
Expand All @@ -263,6 +268,7 @@ private void dropBacklogForTimeLimit(PersistentTopic persistentTopic, BacklogQuo
Position nextPosition = PositionFactory.create(ledgerId, -1);
if (!nextPosition.equals(oldestPosition)) {
slowestConsumer.markDelete(nextPosition);
markDeletePositionMoveForward(persistentTopic, slowestConsumer);
continue;
}
}
Expand Down Expand Up @@ -332,4 +338,48 @@ private boolean advanceSlowestSystemCursor(PersistentTopic persistentTopic) {
// We may need to check other system cursors here : replicator, compaction
return false;
}

/**
* Invoke {@link Dispatcher#markDeletePositionMoveForward()} for the subscription that owns the given cursor.
* This ensures pending acks and redelivery state are cleaned up when the cursor is advanced by
* backlog quota eviction (bypassing the subscription-level wrappers that normally fire this hook).
*
* @param persistentTopic the topic
* @param cursor the cursor that was advanced
*/
private void markDeletePositionMoveForward(PersistentTopic persistentTopic, ManagedCursor cursor) {
PersistentSubscription subscription =
persistentTopic.getSubscriptions().get(Codec.decode(cursor.getName()));
if (subscription != null && subscription.getDispatcher() != null) {
subscription.getDispatcher().markDeletePositionMoveForward();
}
}


/**
* Compute the target value after backlog eviction.
*
* @param quotaLimit configured quota limit
* @return target value after eviction
*/
private static long computeEvictionTarget(long quotaLimit) {
double factor = 0.9;
return (long) (factor * quotaLimit);
}

/**
* Compute the number of entries to evict in a single eviction iteration.
*
* @param currentValue current backlog value
* @param quotaLimit configured quota limit
* @param totalEntries total entries in backlog
* @return entries to evict
*/
@VisibleForTesting
static int computeEntriesToEvict(
long currentValue, long quotaLimit, long totalEntries) {
long evictionTarget = computeEvictionTarget(quotaLimit);
return (int) ((currentValue - evictionTarget)
* (double) totalEntries / currentValue);
}
}
129 changes: 129 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import static java.util.Map.entry;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.pulsar.broker.service.BacklogQuotaManager.computeEntriesToEvict;
import static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertMetricLongGaugeValue;
import static org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil.assertMetricLongSumValue;
import static org.apache.pulsar.common.policies.data.BacklogQuota.BacklogQuotaType.destination_storage;
Expand Down Expand Up @@ -56,6 +57,7 @@
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.stats.BrokerOpenTelemetryTestUtil;
import org.apache.pulsar.broker.stats.OpenTelemetryTopicStats;
Expand All @@ -72,6 +74,7 @@
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.ClusterData;
Expand Down Expand Up @@ -2200,4 +2203,130 @@ public void testBacklogQuotaInGB(boolean backlogQuotaSizeGB) throws Exception {
TopicStats stats = getTopicStats(topic1);
assertTrue(stats.getBacklogSize() < 10 * 1024, "Storage size is [" + stats.getStorageSize() + "]");
}

private void assertPendingAcks(org.apache.pulsar.broker.service.Consumer consumer, int expected) {
PendingAcksMap pendingAcks = consumer.getPendingAcks();
assertThat(pendingAcks).isNotNull();
assertThat(pendingAcks.size()).isEqualTo(expected);
assertThat(consumer.getUnackedMessages()).isEqualTo(expected);
}

@Test
public void testConsumerBacklogEvictionSizeQuotaCleansPendingAcks() throws Exception {
final int msgSize = 1024;
final int quotaSizeLimit = 10 * 1024;
final int numMsgs = 20;

admin.namespaces().setBacklogQuota("prop/ns-quota",
BacklogQuota.builder()
.limitSize(quotaSizeLimit)
.retentionPolicy(BacklogQuota.RetentionPolicy.consumer_backlog_eviction)
.build());

@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(adminUrl.toString())
.build();

final String topic =
BrokerTestUtil.newUniqueName("persistent://prop/ns-quota/topic-pending-acks-size");
final String subName = "key-shared-sub";

@Cleanup
Consumer<byte[]> consumer = client.newConsumer()
.topic(topic)
.subscriptionName(subName)
.subscriptionType(SubscriptionType.Key_Shared)
.subscribe();

@Cleanup
Producer<byte[]> producer = createProducer(client, topic);

byte[] content = new byte[msgSize];
for (int i = 0; i < numMsgs; i++) {
producer.send(content);
}

// Receive all messages but don't ack — pending acks accumulate.
for (int i = 0; i < numMsgs; i++) {
consumer.receive();
}

PersistentTopic topicRef =
(PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
PersistentSubscription sub = topicRef.getSubscription(subName);

org.apache.pulsar.broker.service.Consumer brokerConsumer = sub.getDispatcher().getConsumers().get(0);
assertThat(sub).isNotNull();
assertPendingAcks(brokerConsumer, numMsgs);

int expectedRemaining = numMsgs - computeEntriesToEvict(
(long) numMsgs * msgSize,
quotaSizeLimit,
numMsgs);

Awaitility.await()
.pollDelay(TIME_TO_CHECK_BACKLOG_QUOTA + 1, SECONDS)
.pollInterval(1, SECONDS)
.untilAsserted(() -> assertPendingAcks(brokerConsumer, expectedRemaining));
}

@Test
public void testConsumerBacklogEvictionTimeQuotaNotPreciseCleansPendingAcks()
throws Exception {
admin.namespaces().setBacklogQuota("prop/ns-quota",
BacklogQuota.builder()
.limitTime(TIME_TO_CHECK_BACKLOG_QUOTA)
.retentionPolicy(BacklogQuota.RetentionPolicy.consumer_backlog_eviction)
.build(), message_age);

@Cleanup
PulsarClient client = PulsarClient.builder()
.serviceUrl(adminUrl.toString())
.build();

final String topic =
BrokerTestUtil.newUniqueName("persistent://prop/ns-quota/topic-pending-acks-time");
final String subName = "key-shared-sub-time";
final int numMsgs = 14;

@Cleanup
Consumer<byte[]> consumer = client.newConsumer()
.topic(topic)
.subscriptionName(subName)
.subscriptionType(SubscriptionType.Key_Shared)
.subscribe();

@Cleanup
Producer<byte[]> producer = createProducer(client, topic);

byte[] content = new byte[1024];
for (int i = 0; i < numMsgs; i++) {
producer.send(content);
}

// Receive all messages but don't ack — pending acks accumulate.
for (int i = 0; i < numMsgs; i++) {
consumer.receive();
}

PersistentTopic topicRef =
(PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).get();
PersistentSubscription sub = topicRef.getSubscription(subName);

org.apache.pulsar.broker.service.Consumer brokerConsumer = sub.getDispatcher().getConsumers().get(0);
assertThat(sub).isNotNull();

assertPendingAcks(brokerConsumer, numMsgs);

// Non-precise eviction removes whole closed ledgers only.
// With MAX_ENTRIES_PER_LEDGER=5 and 14 entries:
// ledgers are [5, 5, 4]. The last ledger remains open and is not evicted.
int expectedRemaining = numMsgs % MAX_ENTRIES_PER_LEDGER;

Awaitility.await()
.pollDelay(TIME_TO_CHECK_BACKLOG_QUOTA * 2, SECONDS)
.pollInterval(1, SECONDS)
.untilAsserted(() -> assertPendingAcks(brokerConsumer, expectedRemaining));
}
}
Loading