This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.7 by this push:
new 7537376 Fixed inconsistent behavior for Namespace bundles cache
(#11346)
7537376 is described below
commit 75373769d681d2f15954776c0f9d2afd1e92c869
Author: Matteo Merli <[email protected]>
AuthorDate: Sun Jul 18 07:31:50 2021 +0200
Fixed inconsistent behavior for Namespace bundles cache (#11346)
---
.../common/naming/NamespaceBundleFactory.java | 2 +
.../BrokerServiceBundlesCacheInvalidationTest.java | 91 ++++++++++++++++++++++
2 files changed, 93 insertions(+)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
index e723475..db2b0ae 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
@@ -148,6 +148,8 @@ public class NamespaceBundleFactory implements
ZooKeeperCacheListener<LocalPolic
}
public void invalidateBundleCache(NamespaceName namespace) {
+ pulsar.getLocalZkCacheService().policiesCache().invalidate(
+ AdminResource.joinPath(LOCAL_POLICIES_ROOT,
namespace.toString()));
bundlesCache.synchronous().invalidate(namespace);
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java
new file mode 100644
index 0000000..f70a21b
--- /dev/null
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceBundlesCacheInvalidationTest.java
@@ -0,0 +1,91 @@
+/**
+ * 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;
+
+import static
org.apache.pulsar.broker.service.BrokerService.BROKER_SERVICE_CONFIGURATION_PATH;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.fail;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.ConsumerImpl;
+import org.apache.pulsar.common.policies.data.BundlesData;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class BrokerServiceBundlesCacheInvalidationTest extends BrokerTestBase {
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ super.baseSetup();
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testRecreateNamespace() throws Exception {
+ String namespace = "prop/test-" + System.nanoTime();
+ String topic = namespace + "/my-topic";
+
+ // First create namespace with 20 bundles
+ admin.namespaces().createNamespace(namespace, 20);
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topic)
+ .create();
+
+ producer.send("Hello");
+ producer.close();
+
+ // Force delete and recreate with 32 bundles
+ admin.namespaces().deleteNamespace(namespace, true);
+ admin.namespaces().createNamespace(namespace, 32);
+
+ BundlesData bundlesData = admin.namespaces().getBundles(namespace);
+ log.info("BUNDLES: {}", admin.namespaces().getBundles(namespace));
+ assertEquals(bundlesData.getNumBundles(), 32);
+ }
+}
