You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2023/08/29 08:03:41 UTC
[pulsar] branch branch-3.0 updated (25f76efee83 -> dd151a5a532)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 25f76efee83 [fix][broker] Make sure all inflight writes have finished before completion of compaction (#21067)
new f68589ea280 [fix] [broker] Producer is blocked on creation because backlog exceeded on topic, when dedup is enabled and no producer is there (#20951)
new dd151a5a532 [fix][broker] Fix potential case cause retention policy not working on topic level (#21041)
The 2 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:
.../pulsar/broker/service/BacklogQuotaManager.java | 28 ++++++++++++++-
.../service/persistent/MessageDeduplication.java | 11 ++++++
.../broker/service/persistent/PersistentTopic.java | 26 +++++++-------
.../broker/service/BacklogQuotaManagerTest.java | 40 ++++++++++++++++++----
.../service/persistent/PersistentTopicTest.java | 31 +++++++++++++++++
5 files changed, 117 insertions(+), 19 deletions(-)
[pulsar] 01/02: [fix] [broker] Producer is blocked on creation because backlog exceeded on topic, when dedup is enabled and no producer is there (#20951)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f68589ea280eea63ee786236d2c2023b7001ecb6
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Wed Aug 23 18:03:29 2023 -0700
[fix] [broker] Producer is blocked on creation because backlog exceeded on topic, when dedup is enabled and no producer is there (#20951)
(cherry picked from commit 30073dbac0e941869b43e090d2682935e8f094e5)
---
.../pulsar/broker/service/BacklogQuotaManager.java | 28 ++++++++++++++-
.../service/persistent/MessageDeduplication.java | 11 ++++++
.../broker/service/persistent/PersistentTopic.java | 4 +++
.../broker/service/BacklogQuotaManagerTest.java | 40 ++++++++++++++++++----
4 files changed, 76 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
index bc2541c802e..6ad1697adfc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java
@@ -103,7 +103,10 @@ public class BacklogQuotaManager {
break;
case producer_exception:
case producer_request_hold:
- disconnectProducers(persistentTopic);
+ if (!advanceSlowestSystemCursor(persistentTopic)) {
+ // The slowest is not a system cursor. Disconnecting producers to put backpressure.
+ disconnectProducers(persistentTopic);
+ }
break;
default:
break;
@@ -268,4 +271,27 @@ public class BacklogQuotaManager {
});
}
+
+ /**
+ * Advances the slowest cursor if that is a system cursor.
+ *
+ * @param persistentTopic
+ * @return true if the slowest cursor is a system cursor
+ */
+ private boolean advanceSlowestSystemCursor(PersistentTopic persistentTopic) {
+
+ ManagedLedgerImpl mLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+ ManagedCursor slowestConsumer = mLedger.getSlowestConsumer();
+ if (slowestConsumer == null) {
+ return false;
+ }
+
+ if (PersistentTopic.isDedupCursorName(slowestConsumer.getName())) {
+ persistentTopic.getMessageDeduplication().takeSnapshot();
+ return true;
+ }
+
+ // We may need to check other system cursors here : replicator, compaction
+ return false;
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
index 490be4a8876..d1b4b74945f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java
@@ -27,6 +27,7 @@ import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenCursorCallback;
@@ -130,6 +131,9 @@ public class MessageDeduplication {
private final String replicatorPrefix;
+
+ private final AtomicBoolean snapshotTaking = new AtomicBoolean(false);
+
public MessageDeduplication(PulsarService pulsar, PersistentTopic topic, ManagedLedger managedLedger) {
this.pulsar = pulsar;
this.topic = topic;
@@ -406,6 +410,11 @@ public class MessageDeduplication {
if (log.isDebugEnabled()) {
log.debug("[{}] Taking snapshot of sequence ids map", topic.getName());
}
+
+ if (!snapshotTaking.compareAndSet(false, true)) {
+ return;
+ }
+
Map<String, Long> snapshot = new TreeMap<>();
highestSequencedPersisted.forEach((producerName, sequenceId) -> {
if (snapshot.size() < maxNumberOfProducers) {
@@ -420,11 +429,13 @@ public class MessageDeduplication {
log.debug("[{}] Stored new deduplication snapshot at {}", topic.getName(), position);
}
lastSnapshotTimestamp = System.currentTimeMillis();
+ snapshotTaking.set(false);
}
@Override
public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
log.warn("[{}] Failed to store new deduplication snapshot at {}", topic.getName(), position);
+ snapshotTaking.set(false);
}
}, null);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 593f741ce39..ffb6828c8c3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -194,6 +194,10 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private final TopicName shadowSourceTopic;
static final String DEDUPLICATION_CURSOR_NAME = "pulsar.dedup";
+
+ public static boolean isDedupCursorName(String name) {
+ return DEDUPLICATION_CURSOR_NAME.equals(name);
+ }
private static final String TOPIC_EPOCH_PROPERTY_NAME = "pulsar.topic.epoch";
private static final double MESSAGE_EXPIRY_THRESHOLD = 1.5;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
index f3463ee121d..0ac5fdaef15 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BacklogQuotaManagerTest.java
@@ -127,8 +127,8 @@ public class BacklogQuotaManagerTest {
config.setManagedLedgerMaxEntriesPerLedger(MAX_ENTRIES_PER_LEDGER);
config.setManagedLedgerMinLedgerRolloverTimeMinutes(0);
config.setAllowAutoTopicCreationType(TopicType.NON_PARTITIONED);
- config.setSystemTopicEnabled(false);
- config.setTopicLevelPoliciesEnabled(false);
+ config.setSystemTopicEnabled(true);
+ config.setTopicLevelPoliciesEnabled(true);
config.setForceDeleteNamespaceAllowed(true);
pulsar = new PulsarService(config);
@@ -1169,8 +1169,13 @@ public class BacklogQuotaManagerTest {
assertTrue(gotException, "backlog exceeded exception did not occur");
}
- @Test
- public void testProducerExceptionAndThenUnblockSizeQuota() throws Exception {
+ @DataProvider(name = "dedupTestSet")
+ public static Object[][] dedupTestSet() {
+ return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } };
+ }
+
+ @Test(dataProvider = "dedupTestSet")
+ public void testProducerExceptionAndThenUnblockSizeQuota(boolean dedupTestSet) throws Exception {
assertEquals(admin.namespaces().getBacklogQuotaMap("prop/quotahold"),
new HashMap<>());
admin.namespaces().setBacklogQuota("prop/quotahold",
@@ -1186,9 +1191,12 @@ public class BacklogQuotaManagerTest {
boolean gotException = false;
Consumer<byte[]> consumer = client.newConsumer().topic(topic1).subscriptionName(subName1).subscribe();
-
byte[] content = new byte[1024];
Producer<byte[]> producer = createProducer(client, topic1);
+
+ admin.topicPolicies().setDeduplicationStatus(topic1, dedupTestSet);
+ Thread.sleep((TIME_TO_CHECK_BACKLOG_QUOTA + 1) * 1000);
+
for (int i = 0; i < 10; i++) {
producer.send(content);
}
@@ -1207,6 +1215,7 @@ public class BacklogQuotaManagerTest {
}
assertTrue(gotException, "backlog exceeded exception did not occur");
+ assertFalse(producer.isConnected());
// now remove backlog and ensure that producer is unblocked;
TopicStats stats = getTopicStats(topic1);
@@ -1223,14 +1232,33 @@ public class BacklogQuotaManagerTest {
Exception sendException = null;
gotException = false;
try {
- for (int i = 0; i < 5; i++) {
+ for (int i = 0; i < 10; i++) {
producer.send(content);
+ Message<?> msg = consumer.receive();
+ consumer.acknowledge(msg);
}
} catch (Exception e) {
gotException = true;
sendException = e;
}
+ Thread.sleep((TIME_TO_CHECK_BACKLOG_QUOTA + 1) * 1000);
assertFalse(gotException, "unable to publish due to " + sendException);
+
+ gotException = false;
+ long lastDisconnectedTimestamp = producer.getLastDisconnectedTimestamp();
+ try {
+ // try to send over backlog quota and make sure it passes
+ producer.send(content);
+ producer.send(content);
+ } catch (PulsarClientException ce) {
+ assertTrue(ce instanceof PulsarClientException.ProducerBlockedQuotaExceededException
+ || ce instanceof PulsarClientException.TimeoutException, ce.getMessage());
+ gotException = true;
+ sendException = ce;
+ }
+ assertFalse(gotException, "unable to publish due to " + sendException);
+ assertEquals(lastDisconnectedTimestamp, producer.getLastDisconnectedTimestamp());
+
}
@Test
[pulsar] 02/02: [fix][broker] Fix potential case cause retention policy not working on topic level (#21041)
Posted by yu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit dd151a5a5323076d2faa733bd96182ff685b56ee
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Tue Aug 22 22:08:09 2023 +0800
[fix][broker] Fix potential case cause retention policy not working on topic level (#21041)
(cherry picked from commit d3a6df3c2a8f99ab3e5b7f6d826533ffd23328dc)
---
.../broker/service/persistent/PersistentTopic.java | 22 +++++++--------
.../service/persistent/PersistentTopicTest.java | 31 ++++++++++++++++++++++
2 files changed, 41 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index ffb6828c8c3..3e3363b8c5b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1535,7 +1535,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
CompletableFuture<Void> checkReplicationAndRetryOnFailure() {
CompletableFuture<Void> result = new CompletableFuture<Void>();
checkReplication().thenAccept(res -> {
- log.info("[{}] Policies updated successfully", topic);
result.complete(null);
}).exceptionally(th -> {
log.error("[{}] Policies update failed {}, scheduled retry in {} seconds", topic, th.getMessage(),
@@ -1555,7 +1554,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return messageDeduplication.checkStatus();
}
- private CompletableFuture<Void> checkPersistencePolicies() {
+ @VisibleForTesting
+ CompletableFuture<Void> checkPersistencePolicies() {
TopicName topicName = TopicName.get(topic);
CompletableFuture<Void> future = new CompletableFuture<>();
brokerService.getManagedLedgerConfig(topicName).thenAccept(config -> {
@@ -3486,16 +3486,14 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
replicators.forEach((name, replicator) -> replicator.updateRateLimiter());
shadowReplicators.forEach((name, replicator) -> replicator.updateRateLimiter());
checkMessageExpiry();
- checkReplicationAndRetryOnFailure();
-
- checkDeduplicationStatus();
-
- preCreateSubscriptionForCompactionIfNeeded();
-
- // update managed ledger config
- checkPersistencePolicies();
- }).exceptionally(e -> {
- Throwable t = e instanceof CompletionException ? e.getCause() : e;
+ })
+ .thenCompose(__ -> checkReplicationAndRetryOnFailure())
+ .thenCompose(__ -> checkDeduplicationStatus())
+ .thenCompose(__ -> preCreateSubscriptionForCompactionIfNeeded())
+ .thenCompose(__ -> checkPersistencePolicies())
+ .thenAccept(__ -> log.info("[{}] Policies updated successfully", topic))
+ .exceptionally(e -> {
+ Throwable t = FutureUtil.unwrapCompletionException(e);
log.error("[{}] update topic policy error: {}", topic, t.getMessage(), t);
return null;
});
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
index 412b8207e34..41704af0b8b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
@@ -60,6 +60,7 @@ import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedger;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
+import org.apache.pulsar.broker.service.TopicPoliciesService;
import org.apache.pulsar.broker.stats.PrometheusMetricsTest;
import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -77,7 +78,9 @@ import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.Policies;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.awaitility.Awaitility;
import org.junit.Assert;
@@ -603,4 +606,32 @@ public class PersistentTopicTest extends BrokerTestBase {
return !topic.getManagedLedger().getCursors().iterator().hasNext();
});
}
+
+ @Test
+ public void testCheckPersistencePolicies() throws Exception {
+ final String myNamespace = "prop/ns";
+ admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
+ final String topic = "persistent://" + myNamespace + "/testConfig" + UUID.randomUUID();
+ conf.setForceDeleteNamespaceAllowed(true);
+ pulsarClient.newProducer().topic(topic).create().close();
+ RetentionPolicies retentionPolicies = new RetentionPolicies(1, 1);
+ PersistentTopic persistentTopic = spy((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(topic).get().get());
+ TopicPoliciesService policiesService = spy(pulsar.getTopicPoliciesService());
+ doReturn(policiesService).when(pulsar).getTopicPoliciesService();
+ TopicPolicies policies = new TopicPolicies();
+ policies.setRetentionPolicies(retentionPolicies);
+ doReturn(policies).when(policiesService).getTopicPoliciesIfExists(TopicName.get(topic));
+ persistentTopic.onUpdate(policies);
+ verify(persistentTopic, times(1)).checkPersistencePolicies();
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionSizeInMB(), 1L);
+ assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionTimeMillis(), TimeUnit.MINUTES.toMillis(1));
+ });
+ // throw exception
+ doReturn(CompletableFuture.failedFuture(new RuntimeException())).when(persistentTopic).checkPersistencePolicies();
+ policies.setRetentionPolicies(new RetentionPolicies(2, 2));
+ persistentTopic.onUpdate(policies);
+ assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionSizeInMB(), 1L);
+ assertEquals(persistentTopic.getManagedLedger().getConfig().getRetentionTimeMillis(), TimeUnit.MINUTES.toMillis(1));
+ }
}