[fix] [admin] set offload threshold should fail if ns policies is read-only (apache#19383)

Author: poorbarcode

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java

@@ -50,7 +50,7 @@ public class NamespaceResources extends BaseResources<Policies> {
     private final PartitionedTopicResources partitionedTopicResources;
     private final MetadataStore configurationStore;
 
-    private static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly";
+    public static final String POLICIES_READONLY_FLAG_PATH = "/admin/flags/policies-readonly";
     private static final String NAMESPACE_BASE_PATH = "/namespace";
     private static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data";
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -2046,7 +2046,7 @@ protected CompletableFuture<Void> internalSetOffloadThresholdInSecondsAsync(long
         CompletableFuture<Void> f = new CompletableFuture<>();
 
         validateNamespacePolicyOperationAsync(namespaceName, PolicyName.OFFLOAD, PolicyOperation.WRITE)
-                .thenApply(v -> validatePoliciesReadOnlyAccessAsync())
+                .thenCompose(v -> validatePoliciesReadOnlyAccessAsync())
                 .thenCompose(v -> updatePoliciesAsync(namespaceName,
                         policies -> {
                             if (policies.offload_policies == null) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiOffloadTest.java

@@ -47,14 +47,17 @@
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import org.apache.bookkeeper.mledger.LedgerOffloader;
 import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.resources.NamespaceResources;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.LongRunningProcessStatus;
 import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException;
@@ -290,6 +293,26 @@ public void testSetNamespaceOffloadPolicies() throws Exception {
         assertEquals(admin.namespaces().getOffloadPolicies(myNamespace), policies);
     }
 
+    @Test
+    public void testSetNamespaceOffloadPoliciesFailByReadOnly() throws Exception {
+        boolean setNsPolicyReadOnlySuccess = false;
+        try {
+            pulsar.getConfigurationMetadataStore().put(NamespaceResources.POLICIES_READONLY_FLAG_PATH, "0".getBytes(),
+                    Optional.empty()).join();
+            setNsPolicyReadOnlySuccess = true;
+            admin.namespaces().setOffloadThresholdInSeconds(myNamespace, 300);
+            fail("set offload threshold should fail when ns policies is readonly");
+        } catch (Exception ex){
+            // ignore.
+        } finally {
+            // cleanup.
+            if (setNsPolicyReadOnlySuccess) {
+                pulsar.getConfigurationMetadataStore().delete(NamespaceResources.POLICIES_READONLY_FLAG_PATH,
+                        Optional.empty()).join();
+            }
+        }
+    }
+
     @Test
     public void testSetTopicOffloadPolicies() throws Exception {
         conf.setManagedLedgerOffloadThresholdInSeconds(100);