You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/10 23:56:30 UTC
(pulsar) 08/18: [fix][broker] Fix dispatcherPauseOnAckStatePersistentEnabled and schemaValidationEnforced returning wrong values when applied=true (#26472)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ff2f5736268391987c08f23b15febfa70278a95f
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Fri Sep 11 01:02:05 2026 +0800
[fix][broker] Fix dispatcherPauseOnAckStatePersistentEnabled and schemaValidationEnforced returning wrong values when applied=true (#26472)
(cherry picked from commit 0bb9c6e6fb89f0ec1ac9a065bdab8996214fe4c2)
---
.../broker/admin/impl/PersistentTopicsBase.java | 38 +++--
.../AdminApiSchemaValidationEnforcedTest.java | 23 +++
.../persistent/AckStatePausePolicyTest.java | 156 +++++++++++++++++++++
.../apache/pulsar/client/admin/TopicPolicies.java | 6 +
4 files changed, 213 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 89484169b332..758357a58feb 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -3516,9 +3516,23 @@ public class PersistentTopicsBase extends AdminResource {
protected CompletableFuture<Boolean> internalGetDispatcherPauseOnAckStatePersistent(boolean applied,
boolean isGlobal) {
return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
- .thenApply(op -> op.map(TopicPolicies::getDispatcherPauseOnAckStatePersistentEnabled)
- .orElse(false));
-}
+ .thenCompose(op -> {
+ Boolean topicPolicy = op.map(TopicPolicies::getDispatcherPauseOnAckStatePersistentEnabled)
+ .orElse(null);
+ if (topicPolicy != null) {
+ return CompletableFuture.completedFuture(topicPolicy);
+ }
+ if (!applied) {
+ return CompletableFuture.completedFuture(false);
+ }
+ return getNamespacePoliciesAsync(namespaceName).thenApply(namespacePolicies -> {
+ Boolean namespacePolicy = namespacePolicies.dispatcherPauseOnAckStatePersistentEnabled;
+ return namespacePolicy == null
+ ? config().isDispatcherPauseOnAckStatePersistentEnabled()
+ : namespacePolicy;
+ });
+ });
+ }
protected CompletableFuture<PersistencePolicies> internalGetPersistence(boolean applied, boolean isGlobal) {
return getTopicPoliciesAsyncWithRetry(topicName, isGlobal)
@@ -5140,15 +5154,19 @@ public class PersistentTopicsBase extends AdminResource {
}
protected CompletableFuture<Boolean> internalGetSchemaValidationEnforced(boolean applied) {
- // Schema validation enforced is typically a local policy
return getTopicPoliciesAsyncWithRetry(topicName)
- .thenApply(op -> op.map(TopicPolicies::getSchemaValidationEnforced).orElseGet(() -> {
- if (applied) {
- boolean namespacePolicy = getNamespacePolicies(namespaceName).schema_validation_enforced;
- return namespacePolicy || pulsar().getConfiguration().isSchemaValidationEnforced();
+ .thenCompose(op -> {
+ Boolean topicPolicy = op.map(TopicPolicies::getSchemaValidationEnforced).orElse(null);
+ boolean brokerPolicy = pulsar().getConfiguration().isSchemaValidationEnforced();
+ if (topicPolicy != null) {
+ return CompletableFuture.completedFuture(applied ? topicPolicy || brokerPolicy : topicPolicy);
}
- return false; // Default if not set and not applied
- }));
+ if (!applied) {
+ return CompletableFuture.completedFuture(false);
+ }
+ return getNamespacePoliciesAsync(namespaceName)
+ .thenApply(policies -> policies.schema_validation_enforced || brokerPolicy);
+ });
}
protected CompletableFuture<Void> internalSetSchemaValidationEnforced(boolean schemaValidationEnforcedToSet) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaValidationEnforcedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaValidationEnforcedTest.java
index 5c7f81892b1b..20b656c9bf78 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaValidationEnforcedTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaValidationEnforcedTest.java
@@ -69,6 +69,29 @@ public class AdminApiSchemaValidationEnforcedTest extends MockedPulsarServiceBas
assertFalse(admin.namespaces().getSchemaValidationEnforced(namespace, false));
}
+ @Test
+ public void testGetTopicSchemaValidationEnforcedAppliedWhenBrokerEnabled() throws Exception {
+ String namespace = "schema-validation-enforced/topicApplied";
+ String topicName = "persistent://" + namespace + "/test";
+ admin.namespaces().createNamespace(namespace);
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().setSchemaValidationEnforced(topicName, false);
+
+ boolean previousValue = conf.isSchemaValidationEnforced();
+ try {
+ conf.setSchemaValidationEnforced(false);
+ admin.namespaces().setSchemaValidationEnforced(namespace, true);
+ assertFalse(admin.topics().getSchemaValidationEnforced(topicName, false));
+ assertFalse(admin.topics().getSchemaValidationEnforced(topicName, true));
+
+ conf.setSchemaValidationEnforced(true);
+ assertFalse(admin.topics().getSchemaValidationEnforced(topicName, false));
+ assertTrue(admin.topics().getSchemaValidationEnforced(topicName, true));
+ } finally {
+ conf.setSchemaValidationEnforced(previousValue);
+ }
+ }
+
@Test
public void testDisableSchemaValidationEnforcedNoSchema() throws Exception {
admin.namespaces().createNamespace("schema-validation-enforced/default-no-schema");
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/AckStatePausePolicyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/AckStatePausePolicyTest.java
new file mode 100644
index 000000000000..269e3546e75a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/AckStatePausePolicyTest.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.service.SharedPulsarBaseTest;
+import org.apache.pulsar.client.admin.TopicPolicies;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/** Checks policy inheritance through the public API and the loaded topic's effective policy. */
+@Test(groups = "broker-impl", singleThreaded = true, timeOut = 30000)
+public class AckStatePausePolicyTest extends SharedPulsarBaseTest {
+ private boolean previousBrokerPolicy;
+
+ @BeforeMethod(alwaysRun = true)
+ public void saveConfiguration() {
+ previousBrokerPolicy = getConfig().isDispatcherPauseOnAckStatePersistentEnabled();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void restoreConfiguration() {
+ getConfig().setDispatcherPauseOnAckStatePersistentEnabled(previousBrokerPolicy);
+ }
+
+ @DataProvider(name = "policyScenarios")
+ public Object[][] policyScenarios() {
+ return new Object[][] {
+ {"all policies unset", false, null, null, false},
+ {"broker enabled", true, null, null, true},
+ {"namespace enabled", false, true, null, true},
+ {"namespace disabled", true, false, null, false},
+ {"topic enabled", false, null, true, true},
+ {"topic disabled", true, true, false, false}
+ };
+ }
+
+ @DataProvider(name = "policyScopes")
+ public Object[][] policyScopes() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "policyScenarios")
+ public void testPolicyInheritance(String scenario, boolean brokerPolicy, Boolean namespacePolicy,
+ Boolean topicPolicy, boolean expected) throws Exception {
+ getConfig().setDispatcherPauseOnAckStatePersistentEnabled(brokerPolicy);
+ if (Boolean.TRUE.equals(namespacePolicy)) {
+ admin.namespaces().setDispatcherPauseOnAckStatePersistent(getNamespace());
+ } else if (Boolean.FALSE.equals(namespacePolicy)) {
+ admin.namespaces().removeDispatcherPauseOnAckStatePersistent(getNamespace());
+ }
+
+ for (boolean isGlobal : new boolean[] {false, true}) {
+ String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ TopicPolicies topicPolicies = admin.topicPolicies(isGlobal);
+ setTopicPolicy(topicPolicies, topicName, topicPolicy);
+
+ PersistentTopic topic = getPersistentTopic(topicName);
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertThat(topic.isDispatcherPauseOnAckStatePersistentEnabled())
+ .as("runtime value for %s in %s scope", scenario, scopeName(isGlobal))
+ .isEqualTo(expected));
+ assertThat(topicPolicies.getDispatcherPauseOnAckStatePersistent(topicName, false)
+ .get(5, TimeUnit.SECONDS))
+ .as("raw value for %s in %s scope", scenario, scopeName(isGlobal))
+ .isEqualTo(Boolean.TRUE.equals(topicPolicy));
+ assertThat(topicPolicies.getDispatcherPauseOnAckStatePersistent(topicName, true)
+ .get(5, TimeUnit.SECONDS))
+ .as("applied value for %s in %s scope", scenario, scopeName(isGlobal))
+ .isEqualTo(expected);
+ }
+ }
+
+ @Test(dataProvider = "policyScopes")
+ public void testExistingTopicPolicyWithUnsetFieldInheritsBroker(boolean isGlobal) throws Exception {
+ getConfig().setDispatcherPauseOnAckStatePersistentEnabled(true);
+ String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+ TopicPolicies topicPolicies = admin.topicPolicies(isGlobal);
+ topicPolicies.setMaxProducersAsync(topicName, 1).get(5, TimeUnit.SECONDS);
+
+ PersistentTopic topic = getPersistentTopic(topicName);
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertThat(topic.isDispatcherPauseOnAckStatePersistentEnabled()).isTrue());
+ assertThat(topicPolicies.getDispatcherPauseOnAckStatePersistent(topicName, false)
+ .get(5, TimeUnit.SECONDS)).isFalse();
+ assertThat(topicPolicies.getDispatcherPauseOnAckStatePersistent(topicName, true)
+ .get(5, TimeUnit.SECONDS))
+ .isTrue();
+ }
+
+ @Test
+ public void testLocalAndGlobalPolicyQueriesRemainIndependent() throws Exception {
+ getConfig().setDispatcherPauseOnAckStatePersistentEnabled(false);
+ String topicName = newTopicName();
+ admin.topics().createNonPartitionedTopic(topicName);
+
+ TopicPolicies globalPolicies = admin.topicPolicies(true);
+ globalPolicies.setDispatcherPauseOnAckStatePersistent(topicName).get(5, TimeUnit.SECONDS);
+ TopicPolicies localPolicies = admin.topicPolicies(false);
+ localPolicies.setDispatcherPauseOnAckStatePersistent(topicName).get(5, TimeUnit.SECONDS);
+ localPolicies.removeDispatcherPauseOnAckStatePersistent(topicName).get(5, TimeUnit.SECONDS);
+
+ PersistentTopic topic = getPersistentTopic(topicName);
+ Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
+ assertThat(topic.isDispatcherPauseOnAckStatePersistentEnabled()).isFalse());
+ assertPolicy(localPolicies, topicName, false);
+ assertPolicy(globalPolicies, topicName, true);
+ }
+
+ private void setTopicPolicy(TopicPolicies topicPolicies, String topicName, Boolean value) throws Exception {
+ if (Boolean.TRUE.equals(value)) {
+ topicPolicies.setDispatcherPauseOnAckStatePersistent(topicName).get(5, TimeUnit.SECONDS);
+ } else if (Boolean.FALSE.equals(value)) {
+ topicPolicies.setDispatcherPauseOnAckStatePersistent(topicName).get(5, TimeUnit.SECONDS);
+ topicPolicies.removeDispatcherPauseOnAckStatePersistent(topicName).get(5, TimeUnit.SECONDS);
+ }
+ }
+
+ private PersistentTopic getPersistentTopic(String topicName) throws Exception {
+ return (PersistentTopic) getTopicIfExists(topicName).get(5, TimeUnit.SECONDS).orElseThrow();
+ }
+
+ private void assertPolicy(TopicPolicies topicPolicies, String topicName, boolean expected) throws Exception {
+ assertThat(topicPolicies.getDispatcherPauseOnAckStatePersistent(topicName, false)
+ .get(5, TimeUnit.SECONDS)).isEqualTo(expected);
+ assertThat(topicPolicies.getDispatcherPauseOnAckStatePersistent(topicName, true)
+ .get(5, TimeUnit.SECONDS)).isEqualTo(expected);
+ }
+
+ private String scopeName(boolean isGlobal) {
+ return isGlobal ? "global" : "local";
+ }
+}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/TopicPolicies.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/TopicPolicies.java
index 69ea350dd1fe..073b1f488df1 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/TopicPolicies.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/TopicPolicies.java
@@ -2005,6 +2005,12 @@ public interface TopicPolicies {
/**
* Get the dispatcherPauseOnAckStatePersistentEnabled policy for a given topic asynchronously.
+ * When {@code applied} is true and the policy is not set for the selected local or global topic-policy scope,
+ * the namespace policy and then the broker configuration are used.
+ *
+ * @param topic topic name
+ * @param applied whether to return the applied policy including namespace and broker inheritance
+ * @return a future that completes with the policy value
*/
CompletableFuture<Boolean> getDispatcherPauseOnAckStatePersistent(String topic, boolean applied);