Skip to content
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

[fix] [broker] fix replicated namespaces filter in filterAndUnloadMatchedNamespaceAsync #23100

Merged
merged 15 commits into from
Jul 31, 2024
Merged

[fix] [broker] fix replicated namespaces filter in filterAndUnloadMatchedNamespaceAsync #23100

Changes from 13 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
debf0ff
fix ns isolation policy for repl namespaces
iosdev747 Jul 29, 2024
81ba58a
Revert "fix ns isolation policy for repl namespaces"
iosdev747 Jul 29, 2024
670f9a2
fix ns isolation policy for repl namespaces
iosdev747 Jul 29, 2024
32da100
fix admin tests
iosdev747 Jul 29, 2024
e58e79e
Merge remote-tracking branch 'origin/master' into fixIsolationPolicy
iosdev747 Jul 29, 2024
00819ec
remove unused import
iosdev747 Jul 29, 2024
a2e37fe
add unload bundle flag in pulsar admin (+CLI) and change default
iosdev747 Jul 29, 2024
ebcd1f1
Merge remote-tracking branch 'upstream/master' into fixNSClustersInPo…
iosdev747 Jul 30, 2024
70cd51d
Revert "add unload bundle flag in pulsar admin (+CLI) and change defa…
iosdev747 Jul 30, 2024
8518315
Revert "fix admin tests"
iosdev747 Jul 30, 2024
7bd978f
revert flag changes
iosdev747 Jul 30, 2024
22de125
get policy post ns filter
iosdev747 Jul 30, 2024
7a6e0d5
Make filterAndUnloadMatchedNamespaceAsync asynchronous
lhotari Jul 30, 2024
b8746a4
add multi broker ns isolation policy test for repl ns
iosdev747 Jul 30, 2024
13bf97a
test cleanup
iosdev747 Jul 30, 2024
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
77 changes: 44 additions & 33 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
Expand Down Expand Up @@ -723,8 +723,8 @@ public void setNamespaceIsolationPolicy(
).thenCompose(nsIsolationPolicies -> {
nsIsolationPolicies.setPolicy(policyName, policyData);
return namespaceIsolationPolicies()
.setIsolationDataAsync(cluster, old -> nsIsolationPolicies.getPolicies());
}).thenCompose(__ -> filterAndUnloadMatchedNamespaceAsync(policyData))
.setIsolationDataAsync(cluster, old -> nsIsolationPolicies.getPolicies());
}).thenCompose(__ -> filterAndUnloadMatchedNamespaceAsync(cluster, policyData))
.thenAccept(__ -> {
log.info("[{}] Successful to update clusters/{}/namespaceIsolationPolicies/{}.",
clientAppId(), cluster, policyName);
Expand Down Expand Up @@ -758,42 +758,53 @@ public void setNamespaceIsolationPolicy(
/**
* Get matched namespaces; call unload for each namespaces.
*/
private CompletableFuture<Void> filterAndUnloadMatchedNamespaceAsync(NamespaceIsolationDataImpl policyData) {
private CompletableFuture<Void> filterAndUnloadMatchedNamespaceAsync(String cluster,
NamespaceIsolationDataImpl policyData) {
PulsarAdmin adminClient;
try {
adminClient = pulsar().getAdminClient();
} catch (PulsarServerException e) {
return FutureUtil.failedFuture(e);
}
return adminClient.tenants().getTenantsAsync()
.thenCompose(tenants -> {
Stream<CompletableFuture<List<String>>> completableFutureStream = tenants.stream()
.map(tenant -> adminClient.namespaces().getNamespacesAsync(tenant));
return FutureUtil.waitForAll(completableFutureStream)
.thenApply(namespaces -> {
// if namespace match any policy regex, add it to ns list to be unload.
return namespaces.stream()
.filter(namespaceName ->
policyData.getNamespaces().stream().anyMatch(namespaceName::matches))
.collect(Collectors.toList());
});
}).thenCompose(shouldUnloadNamespaces -> {
if (CollectionUtils.isEmpty(shouldUnloadNamespaces)) {
return CompletableFuture.completedFuture(null);
}
List<CompletableFuture<Void>> futures = shouldUnloadNamespaces.stream()
.map(namespaceName -> adminClient.namespaces().unloadAsync(namespaceName))
.collect(Collectors.toList());
return FutureUtil.waitForAll(futures)
.thenAccept(__ -> {
try {
// write load info to load manager to make the load happens fast
pulsar().getLoadManager().get().writeLoadReportOnZookeeper(true);
} catch (Exception e) {
log.warn("[{}] Failed to writeLoadReportOnZookeeper.", clientAppId(), e);
}
});
});
// compile regex patterns once
List<Pattern> namespacePatterns = policyData.getNamespaces().stream().map(Pattern::compile).toList();
return adminClient.tenants().getTenantsAsync().thenCompose(tenants -> {
List<CompletableFuture<List<String>>> filteredNamespacesForEachTenant = tenants.stream()
.map(tenant -> adminClient.namespaces().getNamespacesAsync(tenant).thenCompose(namespaces -> {
List<CompletableFuture<String>> namespaceNamesInCluster = namespaces.stream()
.filter(namespaceName -> namespacePatterns.stream()
.anyMatch(pattern -> pattern.matcher(namespaceName).matches()))
.map(namespaceName -> adminClient.namespaces().getPoliciesAsync(namespaceName)
.thenApply(policies -> policies.replication_clusters.contains(cluster)
? namespaceName : null))
.collect(Collectors.toList());
return FutureUtil.waitForAll(namespaceNamesInCluster).thenApply(
__ -> namespaceNamesInCluster.stream()
.map(CompletableFuture::join)
.filter(Objects::nonNull)
.collect(Collectors.toList()));
})).toList();
return FutureUtil.waitForAll(filteredNamespacesForEachTenant)
.thenApply(__ -> filteredNamespacesForEachTenant.stream()
.map(CompletableFuture::join)
.flatMap(List::stream)
.collect(Collectors.toList()));
}).thenCompose(shouldUnloadNamespaces -> {
if (CollectionUtils.isEmpty(shouldUnloadNamespaces)) {
return CompletableFuture.completedFuture(null);
}
List<CompletableFuture<Void>> futures = shouldUnloadNamespaces.stream()
.map(namespaceName -> adminClient.namespaces().unloadAsync(namespaceName))
.collect(Collectors.toList());
return FutureUtil.waitForAll(futures).thenAccept(__ -> {
try {
// write load info to load manager to make the load happens fast
pulsar().getLoadManager().get().writeLoadReportOnZookeeper(true);
} catch (Exception e) {
log.warn("[{}] Failed to writeLoadReportOnZookeeper.", clientAppId(), e);
}
});
});
}

@DELETE
Expand Down
Loading