Skip to content

Commit

Permalink
Fixed inconsistent behavior for Namespace bundles cache (#11346)
Browse files Browse the repository at this point in the history
  • Loading branch information
merlimat committed Jul 18, 2021
1 parent ef61fb3 commit 7537376
Show file tree
Hide file tree
Showing 2 changed files with 93 additions and 0 deletions.
Expand Up @@ -148,6 +148,8 @@ private boolean isOwner(NamespaceBundle nsBundle) {
}

public void invalidateBundleCache(NamespaceName namespace) {
pulsar.getLocalZkCacheService().policiesCache().invalidate(
AdminResource.joinPath(LOCAL_POLICIES_ROOT, namespace.toString()));
bundlesCache.synchronous().invalidate(namespace);
}

Expand Down
@@ -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);
}
}

0 comments on commit 7537376

Please sign in to comment.