Make LocalPolicies immutable to avoid concurrent modify inconsistent.

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

@@ -23,6 +23,7 @@
 import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
 import static org.apache.pulsar.broker.cache.ConfigurationCacheService.RESOURCEGROUPS;
 import static org.apache.pulsar.broker.cache.LocalZooKeeperCacheService.LOCAL_POLICIES_ROOT;
+import static org.apache.pulsar.common.policies.data.Policies.defaultBundle;
 import static org.apache.pulsar.common.policies.data.Policies.getBundles;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
@@ -933,8 +934,13 @@ protected void internalSetBookieAffinityGroup(BookieAffinityGroupData bookieAffi
         String path = joinPath(LOCAL_POLICIES_ROOT, this.namespaceName.toString());
         try {
             getLocalPolicies().setWithCreate(path, (oldPolicies) -> {
-                LocalPolicies localPolicies = oldPolicies.orElse(new LocalPolicies());
-                localPolicies.bookieAffinityGroup = bookieAffinityGroup;
+                LocalPolicies localPolicies = oldPolicies.map(
+                        policies -> new LocalPolicies(policies.bundles,
+                                bookieAffinityGroup,
+                                policies.namespaceAntiAffinityGroup))
+                        .orElseGet(() -> new LocalPolicies(defaultBundle(),
+                                bookieAffinityGroup,
+                                null));
                 log.info("[{}] Successfully updated local-policies configuration: namespace={}, map={}", clientAppId(),
                         namespaceName, localPolicies);
                 return localPolicies;
@@ -1755,11 +1761,13 @@ protected void internalSetNamespaceAntiAffinityGroup(String antiAffinityGroup) {
 
         try {
             String path = joinPath(LOCAL_POLICIES_ROOT, this.namespaceName.toString());
-            getLocalPolicies().setWithCreate(path, (lp)->{
-                LocalPolicies localPolicies = lp.orElse(new LocalPolicies());
-                localPolicies.namespaceAntiAffinityGroup = antiAffinityGroup;
-                return localPolicies;
-            });
+            getLocalPolicies().setWithCreate(path, (lp)->
+                lp.map(policies -> new LocalPolicies(policies.bundles,
+                        policies.bookieAffinityGroup,
+                        antiAffinityGroup))
+                        .orElseGet(() -> new LocalPolicies(defaultBundle(),
+                                null, antiAffinityGroup))
+            );
             log.info("[{}] Successfully updated local-policies configuration: namespace={}, map={}", clientAppId(),
                     namespaceName, antiAffinityGroup);
         } catch (Exception e) {
@@ -1790,10 +1798,10 @@ protected void internalRemoveNamespaceAntiAffinityGroup() {
 
         try {
             final String path = joinPath(LOCAL_POLICIES_ROOT, namespaceName.toString());
-            getLocalPolicies().set(path, (policies)->{
-                policies.namespaceAntiAffinityGroup = null;
-                return policies;
-            });
+            getLocalPolicies().set(path, (policies)->
+                new LocalPolicies(policies.bundles,
+                        policies.bookieAffinityGroup,
+                        null));
             log.info("[{}] Successfully removed anti-affinity group for a namespace={}", clientAppId(), namespaceName);
         } catch (Exception e) {
             log.error("[{}] Failed to remove anti-affinity group for namespace {}", clientAppId(), namespaceName, e);

pulsar-broker/src/main/java/org/apache/pulsar/broker/cache/LocalZooKeeperCacheService.java

@@ -175,8 +175,10 @@ public CompletableFuture<Optional<LocalPolicies>> createPolicies(String path, bo
             configurationCacheService.policiesCache().getAsync(globalPath).thenAccept(policies -> {
                 if (policies.isPresent()) {
                     // Copying global bundles information to local policies
-                    LocalPolicies localPolicies = new LocalPolicies();
-                    localPolicies.bundles = policies.get().bundles;
+                    LocalPolicies localPolicies = new LocalPolicies(policies.get().bundles,
+                            null,
+                            null);
+
                     readFromGlobalFuture.complete(Optional.of(localPolicies));
                 } else {
                     // Policies are not present in global zk

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -930,8 +930,20 @@ private void updateNamespaceBundles(NamespaceName nsname, NamespaceBundles nsBun
         }
 
         long version = nsBundles.getVersion();
-        LocalPolicies local = policies.orElse(new LocalPolicies());
-        local.bundles = getBundlesData(nsBundles);
+
+        LocalPolicies local;
+        BundlesData bundlesData = getBundlesData(nsBundles);
+
+        // object copy to avoid concurrent modify LocalPolicy
+        // cause data not equals nsBundles after serialization.
+        local = policies.map(
+                localPolicies -> new LocalPolicies(bundlesData,
+                        localPolicies.bookieAffinityGroup,
+                        localPolicies.namespaceAntiAffinityGroup))
+                .orElseGet(() -> new LocalPolicies(bundlesData,
+                        null,
+                        null));
+
         byte[] data = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(local);
 
         this.pulsar.getLocalZkCache().getZooKeeper()

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/LocalPolicies.java

@@ -19,37 +19,34 @@
 package org.apache.pulsar.common.policies.data;
 
 import static org.apache.pulsar.common.policies.data.Policies.defaultBundle;
-import com.google.common.base.Objects;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
 
 /**
  * Local policies.
  */
+@ToString
+@EqualsAndHashCode
 public class LocalPolicies {
 
-    public BundlesData bundles;
+    public final BundlesData bundles;
     // bookie affinity group for bookie-isolation
-    public BookieAffinityGroupData bookieAffinityGroup;
+    public final BookieAffinityGroupData bookieAffinityGroup;
     // namespace anti-affinity-group
-    public String namespaceAntiAffinityGroup;
+    public final String namespaceAntiAffinityGroup;
 
     public LocalPolicies() {
         bundles = defaultBundle();
+        bookieAffinityGroup = null;
+        namespaceAntiAffinityGroup = null;
     }
 
-    @Override
-    public int hashCode() {
-        return Objects.hashCode(bundles, bookieAffinityGroup);
+    public LocalPolicies(BundlesData data,
+                         BookieAffinityGroupData bookieAffinityGroup,
+                         String namespaceAntiAffinityGroup) {
+        bundles = data;
+        this.bookieAffinityGroup = bookieAffinityGroup;
+        this.namespaceAntiAffinityGroup = namespaceAntiAffinityGroup;
     }
 
-    @Override
-    public boolean equals(Object obj) {
-        if (obj instanceof LocalPolicies) {
-            LocalPolicies other = (LocalPolicies) obj;
-            return Objects.equal(bundles, other.bundles)
-                    && Objects.equal(bookieAffinityGroup, other.bookieAffinityGroup)
-                    && Objects.equal(namespaceAntiAffinityGroup, other.namespaceAntiAffinityGroup);
-        }
-        return false;
-    }
-
-}
+}

pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/LocalPolicesTest.java

@@ -18,12 +18,17 @@
  */
 package org.apache.pulsar.common.policies.data;
 
+import static org.apache.pulsar.common.policies.data.Policies.defaultBundle;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotEquals;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
+import com.google.common.base.Objects;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
 public class LocalPolicesTest {
@@ -49,4 +54,100 @@ public void testLocalPolices() {
         localPolicy1.bundles.setNumBundles(boundaries0.size() - 1);
         assertEquals(localPolicy1, localPolicy0);
     }
+
+    // only for test
+    class MutableLocalPolicies {
+
+        public BundlesData bundles;
+        // bookie affinity group for bookie-isolation
+        public BookieAffinityGroupData bookieAffinityGroup;
+        // namespace anti-affinity-group
+        public String namespaceAntiAffinityGroup;
+
+        public MutableLocalPolicies() {
+            bundles = defaultBundle();
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hashCode(bundles, bookieAffinityGroup);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj instanceof LocalPolicies) {
+                LocalPolicies other = (LocalPolicies) obj;
+                return Objects.equal(bundles, other.bundles)
+                        && Objects.equal(bookieAffinityGroup, other.bookieAffinityGroup)
+                        && Objects.equal(namespaceAntiAffinityGroup, other.namespaceAntiAffinityGroup);
+            }
+            return false;
+        }
+
+    }
+
+    // https://github.com/apache/pulsar/pull/9598
+    @Test
+    public void testMakeLocalPoliciesImmutableSerializationCompatibility() throws IOException {
+
+        // no fields
+        MutableLocalPolicies mutableLocalPolicies = new MutableLocalPolicies();
+        LocalPolicies immutableLocalPolicies = new LocalPolicies();
+
+        // serialize and deserialize
+        byte[] data = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(mutableLocalPolicies);
+        LocalPolicies mutableDeserializedPolicies = ObjectMapperFactory.getThreadLocal().readValue(data, LocalPolicies.class);
+
+        Assert.assertEquals(mutableDeserializedPolicies,immutableLocalPolicies);
+
+
+
+
+        // check with set other fields
+        BookieAffinityGroupData bookieAffinityGroupData = new BookieAffinityGroupData("aaa","bbb");
+        String namespaceAntiAffinityGroup = "namespace1,namespace2";
+
+        mutableLocalPolicies.bookieAffinityGroup = bookieAffinityGroupData;
+        mutableLocalPolicies.namespaceAntiAffinityGroup = namespaceAntiAffinityGroup;
+        LocalPolicies immutableLocalPolicies2 = new LocalPolicies(defaultBundle(),bookieAffinityGroupData,namespaceAntiAffinityGroup);
+
+        // serialize and deserialize
+        data = ObjectMapperFactory.getThreadLocal().writeValueAsBytes(mutableLocalPolicies);
+        mutableDeserializedPolicies = ObjectMapperFactory.getThreadLocal().readValue(data, LocalPolicies.class);
+
+        Assert.assertEquals(mutableDeserializedPolicies,immutableLocalPolicies2);
+
+    }
+
+    @Test
+    public void testMakeLocalPoliciesImmutableStringSerializationCompatibility() throws IOException {
+
+        // no fields
+        MutableLocalPolicies mutableLocalPolicies = new MutableLocalPolicies();
+        LocalPolicies immutableLocalPolicies = new LocalPolicies();
+
+        // serialize and deserialize
+        String data = ObjectMapperFactory.getThreadLocal().writeValueAsString(mutableLocalPolicies);
+        LocalPolicies mutableDeserializedPolicies = ObjectMapperFactory.getThreadLocal().readValue(data, LocalPolicies.class);
+
+        Assert.assertEquals(mutableDeserializedPolicies,immutableLocalPolicies);
+
+
+
+
+        // check with set other fields
+        BookieAffinityGroupData bookieAffinityGroupData = new BookieAffinityGroupData("aaa","bbb");
+        String namespaceAntiAffinityGroup = "namespace1,namespace2";
+
+        mutableLocalPolicies.bookieAffinityGroup = bookieAffinityGroupData;
+        mutableLocalPolicies.namespaceAntiAffinityGroup = namespaceAntiAffinityGroup;
+        LocalPolicies immutableLocalPolicies2 = new LocalPolicies(defaultBundle(),bookieAffinityGroupData,namespaceAntiAffinityGroup);
+
+        // serialize and deserialize
+        data = ObjectMapperFactory.getThreadLocal().writeValueAsString(mutableLocalPolicies);
+        mutableDeserializedPolicies = ObjectMapperFactory.getThreadLocal().readValue(data, LocalPolicies.class);
+
+        Assert.assertEquals(mutableDeserializedPolicies,immutableLocalPolicies2);
+
+    }
 }