You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/07/24 18:28:39 UTC
(pulsar) branch branch-3.3 updated (a75455a73d3 -> 4f63a975fe2)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from a75455a73d3 [improve][test] Remove EntryCacheCreator from ManagedLedgerFactoryImpl (#24552)
new 12fe9f35be3 [fix][test][branch-3.3] Fix ExtensibleLoadManager related tests blocking release
new 4f63a975fe2 [fix][ml] Fix the possibility of message loss or disorder when ML PayloadProcessor processing fails (#24522)
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:
.../apache/bookkeeper/mledger/ManagedLedger.java | 17 +++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 11 ++
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 18 ++-
.../pulsar/broker/loadbalance/LoadManager.java | 4 +
.../SystemTopicBasedTopicPoliciesService.java | 7 ++
.../broker/service/persistent/PersistentTopic.java | 1 +
.../broker/ManagedLedgerPayloadProcessor0.java | 57 +++++++++
.../broker/PublishWithMLPayloadProcessorTest.java | 131 +++++++++++++++++++++
.../ManagedLedgerInterceptorImplTest.java | 45 ++++---
.../extensions/ExtensibleLoadManagerCloseTest.java | 2 +-
.../service/persistent/PersistentTopicTest.java | 3 +-
11 files changed, 275 insertions(+), 21 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java
(pulsar) 02/02: [fix][ml] Fix the possibility of message loss or disorder when ML PayloadProcessor processing fails (#24522)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4f63a975fe2a2b38082ca31ff319cf1203891b90
Author: 道君- Tao Jiuming <da...@apache.org>
AuthorDate: Thu Jul 24 18:16:17 2025 +0800
[fix][ml] Fix the possibility of message loss or disorder when ML PayloadProcessor processing fails (#24522)
Co-authored-by: Yunze Xu <xy...@163.com>
(cherry picked from commit a14794a2a73142f5cbd4fe13b8fca18532ebd001)
---
.../apache/bookkeeper/mledger/ManagedLedger.java | 17 +++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 11 ++
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 18 ++-
.../broker/service/persistent/PersistentTopic.java | 1 +
.../broker/ManagedLedgerPayloadProcessor0.java | 57 +++++++++
.../broker/PublishWithMLPayloadProcessorTest.java | 131 +++++++++++++++++++++
.../ManagedLedgerInterceptorImplTest.java | 45 ++++---
7 files changed, 261 insertions(+), 19 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
index c9507e38065..d016d286800 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
@@ -417,6 +417,23 @@ public interface ManagedLedger {
*/
long getOffloadedSize();
+ /**
+ * Resets the exception thrown by the PayloadProcessor during an add entry operation to null.
+ * <p>
+ * **Context:** When an add entry operation fails due to an interceptor, all subsequent incoming add entry
+ * operations will also fail. This behavior ensures message ordering and consistency.
+ * <p>
+ * **Important:** This method MUST only be called after all pending add operations are fully completed
+ * (e.g., after a Topic is unfenced). Calling it prematurely will prevent the Managed Ledger (ML)
+ * from being able to write indefinitely.
+ * <p>
+ * **Implementation Note:** Downstream projects that support the ML PayloadProcessor should implement
+ * this method. Otherwise, do not implement it.
+ */
+ default void unfenceForInterceptorException() {
+ // Default implementation does nothing
+ }
+
/**
* Get last offloaded ledgerId. If no offloaded yet, it returns 0.
*
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 0040e7b6298..f719d65201e 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
@@ -230,6 +230,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
private volatile long lastOffloadSuccessTimestamp = 0;
private volatile long lastOffloadFailureTimestamp = 0;
+ protected volatile ManagedLedgerException interceptorException = null;
+
private int minBacklogCursorsForCaching = 0;
private int minBacklogEntriesForCaching = 1000;
private int maxBacklogBetweenCursorsForCaching = 1000;
@@ -4324,6 +4326,15 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
return offloadedSize;
}
+ @Override
+ public void unfenceForInterceptorException() {
+ this.interceptorException = null;
+ }
+
+ protected void fenceForInterceptorException(ManagedLedgerException e) {
+ this.interceptorException = e;
+ }
+
@Override
public long getLastOffloadedLedgerId() {
return lastOffloadLedgerId;
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 726a463439f..13b67abc5d0 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -131,8 +131,16 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
public void initiate() {
if (STATE_UPDATER.compareAndSet(OpAddEntry.this, State.OPEN, State.INITIATED)) {
- ByteBuf duplicateBuffer = data.retainedDuplicate();
+ // Fail the add operation if the managed ledger is in a state that prevents adding entries.
+ ManagedLedgerException exbw = ml.interceptorException;
+ if (exbw != null) {
+ ml.pendingAddEntries.remove(this);
+ this.failed(exbw);
+ // Don't recycle the object here, see: https://lists.apache.org/thread/po08w0tkhc7q8gc5khpdft6stxnr1v2y
+ return;
+ }
+ ByteBuf duplicateBuffer = data.retainedDuplicate();
// internally asyncAddEntry() will take the ownership of the buffer and release it at the end
lastInitTime = System.nanoTime();
if (ml.getManagedLedgerInterceptor() != null) {
@@ -141,10 +149,14 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
payloadProcessorHandle = ml.getManagedLedgerInterceptor()
.processPayloadBeforeLedgerWrite(this, duplicateBuffer);
} catch (Exception e) {
+ ManagedLedgerException mle = new ManagedLedgerException.ManagedLedgerInterceptException(e);
+ ml.fenceForInterceptorException(mle);
ml.pendingAddEntries.remove(this);
ReferenceCountUtil.safeRelease(duplicateBuffer);
log.error("[{}] Error processing payload before ledger write", ml.getName(), e);
- this.failed(new ManagedLedgerException.ManagedLedgerInterceptException(e));
+ this.failed(mle);
+ // Don't recycle the object here
+ // see: https://lists.apache.org/thread/po08w0tkhc7q8gc5khpdft6stxnr1v2y
return;
}
if (payloadProcessorHandle != null) {
@@ -169,7 +181,7 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable {
//Use entryId in PublishContext and call addComplete directly.
this.addComplete(BKException.Code.OK, ledger, ((Position) ctx).getEntryId(), addOpCount);
} else {
- log.warn("[{}] initiate with unexpected state {}, expect OPEN state.", ml.getName(), state);
+ log.warn("[{}] initiateShadowWrite with unexpected state {}, expect OPEN state.", ml.getName(), state);
}
}
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 e44e05f0024..2b7221600e9 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
@@ -4315,6 +4315,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
private synchronized void unfence() {
isFenced = false;
+ ledger.unfenceForInterceptorException();
cancelFencedTopicMonitoringTask();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java
new file mode 100644
index 00000000000..e48a0c7e3bb
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/ManagedLedgerPayloadProcessor0.java
@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
+
+@SuppressWarnings("unused") // Used by PublishWithMLPayloadProcessorTest
+public class ManagedLedgerPayloadProcessor0 implements ManagedLedgerPayloadProcessor {
+
+ private final AtomicInteger counter = new AtomicInteger(0);
+ private final int failAt = 4;
+
+ @Override
+ public Processor inputProcessor() {
+ return new Processor() {
+ @Override
+ public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ if (counter.incrementAndGet() == failAt) {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ throw new RuntimeException("Failed to process input payload");
+ }
+ byte[] bytes = new byte[inputPayload.readableBytes()];
+ inputPayload.readBytes(bytes);
+ inputPayload.release();
+ return Unpooled.wrappedBuffer(bytes, 0, bytes.length).retainedDuplicate();
+ }
+
+ @Override
+ public void release(ByteBuf processedPayload) {
+ processedPayload.release();
+ }
+ };
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java
new file mode 100644
index 00000000000..cafd220b6a6
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/PublishWithMLPayloadProcessorTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker")
+public class PublishWithMLPayloadProcessorTest extends ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ conf.setBrokerEntryPayloadProcessors(
+ Collections.singleton(ManagedLedgerPayloadProcessor0.class.getName()));
+ super.internalSetup();
+ super.producerBaseSetup();
+ admin.tenants().createTenant("my-test-tenant", createDefaultTenantInfo());
+ admin.namespaces().createNamespace("my-test-tenant/my-test-ns");
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+
+ @Test(timeOut = 30_000)
+ public void testPublishWithoutDeduplication() throws Exception {
+ String topic = "persistent://my-test-tenant/my-test-ns/testPublishWithoutDeduplication";
+ admin.topics().createNonPartitionedTopic(topic);
+ try {
+ admin.topicPolicies().setDeduplicationStatus(topic, false);
+ publishAndVerify(topic, false);
+ } finally {
+ admin.topics().delete(topic);
+ }
+ }
+
+ @Test(timeOut = 30_000)
+ public void testPublishWithDeduplication() throws Exception {
+ String topic = "persistent://my-test-tenant/my-test-ns/testPublishWithDeduplication";
+ admin.topics().createNonPartitionedTopic(topic);
+ try {
+ admin.topicPolicies().setDeduplicationStatus(topic, true);
+ publishAndVerify(topic, true);
+ } finally {
+ admin.topics().delete(topic);
+ }
+ }
+
+
+ private void publishAndVerify(String topic, boolean enableDeduplication) throws Exception {
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+
+ CountDownLatch latch = new CountDownLatch(10);
+ for (int i = 0; i < 10; i++) {
+ producer.sendAsync("message-" + i).whenComplete((ignored, e) -> {
+ if (e != null) {
+ log.error("Failed to publish message", e);
+ }
+ latch.countDown();
+ });
+ }
+
+ latch.await();
+
+ @Cleanup
+ Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic).subscriptionName("my-sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+
+ List<String> messageContents = new ArrayList<>();
+
+ int messageCount = 0;
+ for (;;) {
+ var msg = consumer.receive(3, TimeUnit.SECONDS);
+ if (msg == null) {
+ break;
+ }
+ String content = msg.getValue();
+ messageCount++;
+ if (!messageContents.contains(content)) {
+ messageContents.add(content);
+ }
+ consumer.acknowledge(msg);
+ }
+
+ if (enableDeduplication) {
+ Assert.assertEquals(messageCount, 10, "Expected 10 total messages");
+ }
+ Assert.assertEquals(messageContents.size(), 10);
+
+ for (int i = 0; i < 10; i++) {
+ Assert.assertEquals(messageContents.get(i), "message-" + i);
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
index 563e17494f4..ef2d3d2fc9e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImplTest.java
@@ -454,6 +454,7 @@ public class ManagedLedgerInterceptorImplTest extends MockedBookKeeperTestCase
return new Processor() {
@Override
public ByteBuf process(Object contextObj, ByteBuf inputPayload) {
+ log.info("testManagedLedgerPayloadInputProcessorFailure.process");
if (inputPayload.readBoolean()) {
throw new RuntimeException(failureMsg);
}
@@ -474,26 +475,38 @@ public class ManagedLedgerInterceptorImplTest extends MockedBookKeeperTestCase
var successCount = new AtomicInteger(0);
var expectedException = new ArrayList<Exception>();
- var addEntryCallback = new AsyncCallbacks.AddEntryCallback() {
- @Override
- public void addComplete(Position position, ByteBuf entryData, Object ctx) {
- successCount.incrementAndGet();
- countDownLatch.countDown();
- }
-
- @Override
- public void addFailed(ManagedLedgerException exception, Object ctx) {
- // expected
- expectedException.add(exception);
- countDownLatch.countDown();
- }
- };
+ ByteBuf shouldFail = Unpooled.buffer().writeBoolean(true);
+ ByteBuf shouldSucceed = Unpooled.buffer().writeBoolean(false);
+ byte[] shouldFailBytes = new byte[shouldFail.readableBytes()];
+ shouldFail.readBytes(shouldFailBytes);
+ byte[] shouldSucceedBytes = new byte[shouldSucceed.readableBytes()];
+ shouldSucceed.readBytes(shouldSucceedBytes);
+ shouldSucceed.release();
+ shouldFail.release();
for (int i = 0; i < count; i++) {
if (i % 2 == 0) {
- ledger.asyncAddEntry(Unpooled.buffer().writeBoolean(true), addEntryCallback, null);
+ try {
+ ledger.addEntry(shouldFailBytes);
+ successCount.incrementAndGet();
+ countDownLatch.countDown();
+ } catch (Exception t) {
+ expectedException.add(t);
+ countDownLatch.countDown();
+ } finally {
+ ledger.unfenceForInterceptorException();
+ }
} else {
- ledger.asyncAddEntry(Unpooled.buffer().writeBoolean(false), addEntryCallback, null);
+ try {
+ ledger.addEntry(shouldSucceedBytes);
+ successCount.incrementAndGet();
+ countDownLatch.countDown();
+ } catch (Exception t) {
+ expectedException.add(t);
+ countDownLatch.countDown();
+ } finally {
+ ledger.unfenceForInterceptorException();
+ }
}
}
(pulsar) 01/02: [fix][test][branch-3.3] Fix ExtensibleLoadManager related tests blocking release
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 12fe9f35be32d7c9becec3567e369a295093b488
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Thu Jul 24 21:12:41 2025 +0300
[fix][test][branch-3.3] Fix ExtensibleLoadManager related tests blocking release
backports #24451 commit 91a40e2 from branch-3.0 to branch-3.3
---
.../java/org/apache/pulsar/broker/loadbalance/LoadManager.java | 4 ++++
.../broker/service/SystemTopicBasedTopicPoliciesService.java | 7 +++++++
.../loadbalance/extensions/ExtensibleLoadManagerCloseTest.java | 2 +-
.../pulsar/broker/service/persistent/PersistentTopicTest.java | 3 ++-
4 files changed, 14 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
index 0dd5d948480..db2fb2ffd0f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/LoadManager.java
@@ -53,6 +53,10 @@ public interface LoadManager {
void start() throws PulsarServerException;
+ default boolean started() {
+ return true;
+ }
+
/**
* Is centralized decision making to assign a new bundle.
*/
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
index 902283b4532..6ea557f63c9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
@@ -336,6 +336,13 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
if (NamespaceService.isHeartbeatNamespace(namespace) || isSelf(topicName)) {
return CompletableFuture.completedFuture(Optional.empty());
}
+ // When the extensible load manager initializes its channel topic, it will trigger the topic policies
+ // initialization by calling this method. At the moment, the load manager does not start so the lookup
+ // for "__change_events" will fail. In this case, just return an empty policies to avoid deadlock.
+ final var loadManager = pulsarService.getLoadManager().get();
+ if (loadManager == null || !loadManager.started() || closed.get()) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
final CompletableFuture<Void> preparedFuture = prepareInitPoliciesCacheAsync(topicName.getNamespaceObject());
return preparedFuture.thenApply(__ -> {
final TopicPolicies candidatePolicies = isGlobal
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerCloseTest.java
index c8427d1a66d..d346ae53e1b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerCloseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerCloseTest.java
@@ -38,7 +38,7 @@ import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@Slf4j
-@Test(groups = "broker")
+@Test(groups = "flaky")
public class ExtensibleLoadManagerCloseTest {
private static final String clusterName = "test";
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 6bbd598b454..aa2d177a4ad 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
@@ -735,7 +735,8 @@ public class PersistentTopicTest extends BrokerTestBase {
doReturn(policiesService).when(pulsar).getTopicPoliciesService();
TopicPolicies policies = new TopicPolicies();
policies.setRetentionPolicies(retentionPolicies);
- doReturn(CompletableFuture.completedFuture(Optional.of(policies))).when(policiesService).getTopicPoliciesAsync(TopicName.get(topic));
+ doReturn(CompletableFuture.completedFuture(Optional.of(policies))).when(policiesService)
+ .getTopicPoliciesAsync(TopicName.get(topic), false);
persistentTopic.onUpdate(policies);
verify(persistentTopic, times(1)).checkPersistencePolicies();
Awaitility.await().untilAsserted(() -> {