You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/04/15 16:19:28 UTC
(pulsar) 02/11: Revert "[fix][admin] Refactor namespace migration operation to async in rest api (#25478)"
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eefa9f0fc049b8ff225c3fa56e77434f44217cbf
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Apr 15 17:34:20 2026 +0300
Revert "[fix][admin] Refactor namespace migration operation to async in rest api (#25478)"
This reverts commit 066a3f30a69c1e6c33902ed4716bc4d8d6bcc2e9.
---
.../pulsar/broker/admin/impl/NamespacesBase.java | 38 ++++++--
.../apache/pulsar/broker/admin/v2/Namespaces.java | 12 +--
.../pulsar/broker/admin/NamespacesV2Test.java | 102 +--------------------
3 files changed, 33 insertions(+), 119 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
index 3b22ea64571..8e1d11b15f3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
@@ -2847,15 +2847,22 @@ public abstract class NamespacesBase extends AdminResource {
});
}
- protected CompletableFuture<Void> internalEnableMigrationAsync(boolean migrated) {
- return validateSuperUserAccessAsync().thenCompose(__ -> getDefaultBundleDataAsync().thenCompose(
- defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
- oldPolicies -> oldPolicies.map(
- policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- policies.namespaceAntiAffinityGroup, migrated))
- .orElseGet(() -> new LocalPolicies(defaultBundleData, null, null, migrated)))))
- .thenAccept(__ -> log.info("Successfully updated migration on namespace {}", namespaceName));
+ protected void internalEnableMigration(boolean migrated) {
+ validateSuperUserAccess();
+ try {
+ getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, oldPolicies -> oldPolicies.map(
+ policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup,
+ policies.namespaceAntiAffinityGroup,
+ migrated))
+ .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, null, migrated)));
+ log.info("Successfully updated migration on namespace {}", namespaceName);
+ } catch (RestException re) {
+ throw re;
+ } catch (Exception e) {
+ log.error("Failed to update migration on namespace {}", namespaceName, e);
+ throw new RestException(e);
+ }
}
protected Policies getDefaultPolicesIfNull(Policies policies) {
@@ -2953,6 +2960,19 @@ public abstract class NamespacesBase extends AdminResource {
.thenApply(policies -> policies.allowed_clusters);
}
+ // TODO remove this sync method after async refactor
+ @Deprecated
+ private BundlesData getDefaultBundleData() {
+ try {
+ return getDefaultBundleDataAsync().get(config().getMetadataStoreOperationTimeoutSeconds(),
+ TimeUnit.SECONDS);
+ } catch (Exception e) {
+ log.error("[{}] Failed to get namespace-policy configuration for namespace {}", clientAppId(),
+ namespaceName, e);
+ throw new RestException(e);
+ }
+ }
+
private CompletableFuture<BundlesData> getDefaultBundleDataAsync() {
return namespaceResources().getPoliciesAsync(namespaceName).thenApply(
optionalPolicies -> optionalPolicies.isPresent() ? optionalPolicies.get().bundles :
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
index 29f652b10f6..a997e1d898a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java
@@ -3066,19 +3066,11 @@ public class Namespaces extends NamespacesBase {
@ApiResponse(code = 200, message = "Operation successful"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
- public void enableMigration(@Suspended AsyncResponse asyncResponse,
- @PathParam("tenant") String tenant,
+ public void enableMigration(@PathParam("tenant") String tenant,
@PathParam("namespace") String namespace,
boolean migrated) {
validateNamespaceName(tenant, namespace);
- internalEnableMigrationAsync(migrated)
- .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
- .exceptionally(ex -> {
- log.error("[{}] Failed to update migration, tenant: {}, namespace: {}",
- clientAppId(), tenant, namespace, ex);
- resumeAsyncResponseExceptionally(asyncResponse, ex);
- return null;
- });
+ internalEnableMigration(migrated);
}
@POST
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
index 9fd1ea1dc01..629e92c056f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java
@@ -24,7 +24,6 @@ import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -353,8 +352,7 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set enable migration
boolean enableMigrationReq = true;
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs,
- enableMigrationReq));
+ namespaces.enableMigration(testTenant, enableMigrationGroupNs, enableMigrationReq);
// 3.query namespace num bundles, should be conf.getDefaultNumberOfNamespaceBundles()
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -378,8 +376,7 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
// 2.set enable migration
boolean enableMigrationReq = true;
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs,
- enableMigrationReq));
+ namespaces.enableMigration(testTenant, enableMigrationGroupNs, enableMigrationReq);
// 3.query namespace num bundles, should be policies.bundles, which we set before
BundlesData bundlesData = (BundlesData) asyncRequests(
@@ -421,99 +418,4 @@ public class NamespacesV2Test extends MockedPulsarServiceBaseTest {
assertNull(bookieAffinityGroupDataResp);
}
- @Test
- public void testSetAndDeleteNamespaceAntiAffinityGroup() throws Exception {
- // 1. create namespace with empty policies, namespace anti affinity group should be null
- String setNamespaceAntiAffinityGroupNs = "test-set-namespace-anti-affinity-group-ns";
- asyncRequests(
- response -> namespaces.createNamespace(response, testTenant, setNamespaceAntiAffinityGroupNs, null));
- String namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
- assertNull(namespaceAntiAffinityGroupResp);
-
- // 2.set namespace anti affinity group
- String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
- asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs, namespaceAntiAffinityGroupReq));
-
- // 3.assert namespace anti affinity group
- namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
- assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
-
- // 4.delete namespace anti affinity group
- asyncRequests(response -> namespaces.removeNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
-
- // 5.assert namespace anti affinity group
- namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant,
- setNamespaceAntiAffinityGroupNs));
- assertNull(namespaceAntiAffinityGroupResp);
- }
-
- @Test
- @SuppressWarnings("unchecked")
- public void testGetClusterAntiAffinityNamespaces() throws Exception {
- // create 5 namespaces, 3 namespaces are set to the same namespace anti affinity group,
- // 2 namespaces are not set to any anti affinity group
- String namespaceWithAntiAffinity1 = "namespace-with-anti-affinity-1";
- String namespaceWithAntiAffinity2 = "namespace-with-anti-affinity-2";
- String namespaceWithAntiAffinity3 = "namespace-with-anti-affinity-3";
- String namespaceWithoutAntiAffinity1 = "namespace-without-anti-affinity-1";
- String namespaceWithoutAntiAffinity2 = "namespace-without-anti-affinity-2";
-
- // create namespaces
- List<String> allNamespaces =
- List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3,
- namespaceWithoutAntiAffinity1, namespaceWithoutAntiAffinity2);
- for (String namespace : allNamespaces) {
- asyncRequests(response -> namespaces.createNamespace(response, testTenant, namespace, null));
- }
-
- // set namespace anti affinity group
- String namespaceAntiAffinityGroupReq = "namespace-anti-affinity-group";
- List<String> namespacesWithAntiAffinityGroup =
- List.of(namespaceWithAntiAffinity1, namespaceWithAntiAffinity2, namespaceWithAntiAffinity3);
- for (String namespace : namespacesWithAntiAffinityGroup) {
- asyncRequests(response -> namespaces.setNamespaceAntiAffinityGroup(response, testTenant, namespace,
- namespaceAntiAffinityGroupReq));
- }
-
- // assert namespace anti affinity group
- for (String namespace : namespacesWithAntiAffinityGroup) {
- String namespaceAntiAffinityGroupResp = (String) asyncRequests(
- response -> namespaces.getNamespaceAntiAffinityGroup(response, testTenant, namespace));
- assertEquals(namespaceAntiAffinityGroupResp, namespaceAntiAffinityGroupReq);
- }
-
- // get namespaces in cluster of given anti affinity group
- List<String> namespacesResp = (List<String>) asyncRequests(
- response -> namespaces.getAntiAffinityNamespaces(response, testLocalCluster,
- namespaceAntiAffinityGroupReq, testTenant));
- List<String> namespacesWithFullPath =
- namespacesWithAntiAffinityGroup.stream().map(ns -> NamespaceName.get(testTenant, ns))
- .map(NamespaceName::toString).toList();
- assertEquals(namespacesResp, namespacesWithFullPath);
- }
-
- @Test
- public void testEnableMigrationAndDisableMigration() throws Exception {
- String enableMigrationGroupNs = "test-enable-migration-disable-migration-ns";
- asyncRequests(response -> namespaces.createNamespace(response, testTenant, enableMigrationGroupNs, null));
-
- // Enable migration
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs, true));
- Policies policiesResp = (Policies) asyncRequests(
- response -> namespaces.getPolicies(response, testTenant, enableMigrationGroupNs));
- assertTrue(policiesResp.migrated);
-
- // Disable migration
- asyncRequests(response -> namespaces.enableMigration(response, testTenant, enableMigrationGroupNs, false));
- policiesResp = (Policies) asyncRequests(
- response -> namespaces.getPolicies(response, testTenant, enableMigrationGroupNs));
- assertFalse(policiesResp.migrated);
- }
}