From 7683f1d2e174ad4e49ad1b659df84f7af7947e86 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sun, 6 Sep 2026 16:12:31 +0800 Subject: [PATCH 1/2] [fix][broker] Honor applied ack-state pause policy inheritance Resolve inherited namespace and broker values when querying the dispatcher pause-on-ack-state-persistence policy with applied=true, while preserving explicit false values and raw-query semantics. Assisted-by: Codex (GPT-5) --- .../admin/impl/PersistentTopicsBase.java | 20 ++- .../persistent/AckStatePausePolicyTest.java | 156 ++++++++++++++++++ .../pulsar/client/admin/TopicPolicies.java | 6 + 3 files changed, 179 insertions(+), 3 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/AckStatePausePolicyTest.java 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 2b39835dbf233..ae4114526a940 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 @@ -3798,9 +3798,23 @@ protected CompletableFuture internalRemoveDispatcherPauseOnAckStatePersist protected CompletableFuture 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; + }); + }); + } @SuppressWarnings("deprecation") protected CompletableFuture internalGetPersistence(boolean applied, boolean isGlobal) { 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 0000000000000..269e3546e75aa --- /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 69ea350dd1fe4..073b1f488df17 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 @@ AutoSubscriptionCreationOverride getAutoSubscriptionCreation(String topic, /** * 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 getDispatcherPauseOnAckStatePersistent(String topic, boolean applied); From ca3f4d4911800463f6d68bffb9d43b59bb5650b7 Mon Sep 17 00:00:00 2001 From: void-ptr974 Date: Sun, 6 Sep 2026 18:11:25 +0800 Subject: [PATCH 2/2] [fix][broker] Fix schemaValidationEnforced applied value Assisted-by: Codex --- .../admin/impl/PersistentTopicsBase.java | 19 ++++++++------- .../AdminApiSchemaValidationEnforcedTest.java | 23 +++++++++++++++++++ 2 files changed, 34 insertions(+), 8 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 ae4114526a940..1359d6a8e9233 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 @@ -5591,17 +5591,20 @@ protected CompletableFuture internalSetSchemaCompatibilityStrategy( }); } - @SuppressWarnings("deprecation") protected CompletableFuture 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 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 c625fb6857848..8e2fc4cee794f 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 void testGetSchemaValidationEnforcedApplied() throws Exception { 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 @SuppressWarnings("unchecked") public void testDisableSchemaValidationEnforcedNoSchema() throws Exception {