Skip to content

[fix] [broker] remove bundle-data in local metadata store. #21078

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
poorbarcode merged 9 commits into from
Aug 31, 2023
Merged

[fix] [broker] remove bundle-data in local metadata store. #21078

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
265bfe7
remove bundle-data in local metadata store.
thetumbled Aug 28, 2023
76dc273
fix checkstyle.
thetumbled Aug 28, 2023
73e86e5
add test code.
thetumbled Aug 28, 2023
23e42c5
fix test code.
thetumbled Aug 28, 2023
e656139
add test code.
thetumbled Aug 29, 2023
8e4473d
add test code.
thetumbled Aug 29, 2023
a9472bc
add test code.
thetumbled Aug 30, 2023
efb87b1
fix checkstyle.
thetumbled Aug 30, 2023
cc9bf78
fix.
thetumbled Aug 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions ...ar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,16 +49,18 @@ public class NamespaceResources extends BaseResources<Policies> {
private final IsolationPolicyResources isolationPolicies;
private final PartitionedTopicResources partitionedTopicResources;
private final MetadataStore configurationStore;
private final MetadataStore localStore;

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";

public NamespaceResources(MetadataStore configurationStore, int operationTimeoutSec) {
public NamespaceResources(MetadataStore localStore, MetadataStore configurationStore, int operationTimeoutSec) {

@AnonHxy AnonHxy Aug 29, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about creating a separate class naming BundleResources to handle the bundle-date creating and deleting. The reason is that:

  1. NamespaceResources will be null if configurationMetadataStore != null is false, so there is a chance to throw NPE when call NamespaceResources.deleteBundleDataAsync(..) or NamespaceResources.deleteBundleDataTenantAsync(...):

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

Lines 60 to 64 in d099ac4

if (configurationMetadataStore != null) {
tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec);
clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec);
namespaceResources = new NamespaceResources(configurationMetadataStore, operationTimeoutSec);
resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec);

  1. NamespaceResources#BUNDLE_DATA_BASE_PATH and ModularLoadManagerImpl#BUNDLE_DATA_PATH are duplicated. BundleResources could unify them. WDYT

@thetumbled thetumbled Aug 29, 2023 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. It seems that we can't allow the configurationMetadataStore to be null, or there will be serious issues.

    pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

    Lines 752 to 762 in d099ac4

    if (config.isConfigurationStoreSeparated()) {
    configMetadataSynchronizer = StringUtils.isNotBlank(config.getConfigurationMetadataSyncEventTopic())
    ? new PulsarMetadataEventSynchronizer(this, config.getConfigurationMetadataSyncEventTopic())
    : null;
    configurationMetadataStore = createConfigurationMetadataStore(configMetadataSynchronizer);
    shouldShutdownConfigurationMetadataStore = true;
    } else {
    configurationMetadataStore = localMetadataStore;
    shouldShutdownConfigurationMetadataStore = false;
    }
    pulsarResources = newPulsarResources();

  2. Currently, bundle-data are created/updated by LoadManager, and deleted by LoadManager and NamespaceResources. It is beneficial to introduce a BundleResources to unify the opeartions to bundle-data. But i think that it should be introduced as a code optimization instead of bug fix, which need to be discussed fully.

@AnonHxy AnonHxy Aug 30, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK. As a bug fix, this change could fix the problem and it LGTM. Maybe we could refactor it in another PR

super(configurationStore, Policies.class, operationTimeoutSec);
this.configurationStore = configurationStore;
isolationPolicies = new IsolationPolicyResources(configurationStore, operationTimeoutSec);
partitionedTopicResources = new PartitionedTopicResources(configurationStore, operationTimeoutSec);
this.localStore = localStore;
}

public CompletableFuture<List<String>> listNamespacesAsync(String tenant) {
Expand Down Expand Up @@ -381,13 +383,13 @@ public CompletableFuture<Void> runWithMarkDeleteAsync(TopicName topic,
// clear resource of `/loadbalance/bundle-data/{tenant}/{namespace}/` in metadata-store
public CompletableFuture<Void> deleteBundleDataAsync(NamespaceName ns) {
final String namespaceBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, ns.toString());
return getStore().deleteRecursive(namespaceBundlePath);
Comment thread
thetumbled marked this conversation as resolved.
return this.localStore.deleteRecursive(namespaceBundlePath);
}

// clear resource of `/loadbalance/bundle-data/{tenant}/` in metadata-store
public CompletableFuture<Void> deleteBundleDataTenantAsync(String tenant) {
final String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant);
return getStore().deleteRecursive(tenantBundlePath);
return this.localStore.deleteRecursive(tenantBundlePath);
}

}
3 changes: 2 additions & 1 deletion pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/PulsarResources.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,8 @@ public PulsarResources(MetadataStore localMetadataStore, MetadataStore configura
if (configurationMetadataStore != null) {
tenantResources = new TenantResources(configurationMetadataStore, operationTimeoutSec);
clusterResources = new ClusterResources(configurationMetadataStore, operationTimeoutSec);
namespaceResources = new NamespaceResources(configurationMetadataStore, operationTimeoutSec);
namespaceResources = new NamespaceResources(localMetadataStore, configurationMetadataStore
, operationTimeoutSec);
resourcegroupResources = new ResourceGroupResources(configurationMetadataStore, operationTimeoutSec);
} else {
tenantResources = null;
Expand Down
44 changes: 44 additions & 0 deletions ...roker-common/src/test/java/org/apache/pulsar/broker/resources/NamespaceResourcesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,17 +18,61 @@
*/
package org.apache.pulsar.broker.resources;

import static org.apache.pulsar.broker.resources.BaseResources.joinPath;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;

import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;


public class NamespaceResourcesTest {

private MetadataStore localStore;
private MetadataStore configurationStore;
private NamespaceResources namespaceResources;

private static final String BUNDLE_DATA_BASE_PATH = "/loadbalance/bundle-data";

@BeforeMethod
public void setup() {
localStore = mock(MetadataStore.class);
configurationStore = mock(MetadataStore.class);
namespaceResources = new NamespaceResources(localStore, configurationStore, 30);
}

@Test
public void test_pathIsFromNamespace() {
assertFalse(NamespaceResources.pathIsFromNamespace("/admin/clusters"));
assertFalse(NamespaceResources.pathIsFromNamespace("/admin/policies"));
assertFalse(NamespaceResources.pathIsFromNamespace("/admin/policies/my-tenant"));
assertTrue(NamespaceResources.pathIsFromNamespace("/admin/policies/my-tenant/my-ns"));
}

/**
* Test that the bundle-data node is deleted from the local stores.
*/
@Test
public void testDeleteBundleDataAsync() {
NamespaceName ns = NamespaceName.get("my-tenant/my-ns");
String namespaceBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, ns.toString());
namespaceResources.deleteBundleDataAsync(ns);

String tenant="my-tenant";
String tenantBundlePath = joinPath(BUNDLE_DATA_BASE_PATH, tenant);
namespaceResources.deleteBundleDataTenantAsync(tenant);

verify(localStore).deleteRecursive(namespaceBundlePath);
verify(localStore).deleteRecursive(tenantBundlePath);

assertThrows(()-> verify(configurationStore).deleteRecursive(namespaceBundlePath));
assertThrows(()-> verify(configurationStore).deleteRecursive(tenantBundlePath));
}


}
31 changes: 26 additions & 5 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl;
import org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper;
import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
Expand Down Expand Up @@ -1720,6 +1721,8 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
// Set conf.
cleanup();
setNamespaceAttr(namespaceAttr);
this.conf.setMetadataStoreUrl("127.0.0.1:2181");
this.conf.setConfigurationMetadataStoreUrl("127.0.0.1:2182");
setup();

String tenant = newUniqueName("test-tenant");
Expand All @@ -1740,6 +1743,28 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
admin.topics().createPartitionedTopic(topic, 10);
assertFalse(admin.topics().getList(namespace).isEmpty());

final String managedLedgersPath = "/managed-ledgers/" + namespace;
final String bundleDataPath = "/loadbalance/bundle-data/" + namespace;
// Trigger bundle owned by brokers.
pulsarClient.newProducer().topic(topic).create().close();
// Trigger bundle data write to ZK.
Awaitility.await().untilAsserted(() -> {
boolean bundleDataWereWriten = false;
for (PulsarService ps : new PulsarService[]{pulsar, mockPulsarSetup.getPulsar()}) {
ModularLoadManagerWrapper loadManager = (ModularLoadManagerWrapper) ps.getLoadManager().get();
ModularLoadManagerImpl loadManagerImpl = (ModularLoadManagerImpl) loadManager.getLoadManager();
ps.getBrokerService().updateRates();
loadManagerImpl.updateLocalBrokerData();
loadManagerImpl.writeBundleDataOnZooKeeper();
bundleDataWereWriten = bundleDataWereWriten || ps.getLocalMetadataStore().exists(bundleDataPath).join();
}
assertTrue(bundleDataWereWriten);
});

// assert znode exists in metadata store
assertTrue(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
assertTrue(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join());

try {
admin.namespaces().deleteNamespace(namespace, false);
fail("should have failed due to namespace not empty");
Expand All @@ -1756,12 +1781,8 @@ public void testDeleteNamespace(NamespaceAttr namespaceAttr) throws Exception {
assertFalse(admin.namespaces().getNamespaces(tenant).contains(namespace));
assertTrue(admin.namespaces().getNamespaces(tenant).isEmpty());


final String managedLedgersPath = "/managed-ledgers/" + namespace;
// assert znode deleted in metadata store
assertFalse(pulsar.getLocalMetadataStore().exists(managedLedgersPath).join());


final String bundleDataPath = "/loadbalance/bundle-data/" + namespace;
assertFalse(pulsar.getLocalMetadataStore().exists(bundleDataPath).join());
}

Expand Down
3 changes: 2 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -715,7 +715,8 @@ protected void initializePulsarServices(SpyConfig spyConfig, Builder builder) {
if (metadataStore == null) {
metadataStore = builder.configurationMetadataStore;
}
NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class, metadataStore, 30);
NamespaceResources nsr = spyConfigPulsarResources.spy(NamespaceResources.class,
builder.localMetadataStore, metadataStore, 30);
TopicResources tsr = spyConfigPulsarResources.spy(TopicResources.class, metadataStore);
pulsarResources(
spyConfigPulsarResources.spy(
Expand Down