You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "oneby-wang (via GitHub)" <gi...@apache.org> on 2025/12/17 00:41:30 UTC
[PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
oneby-wang opened a new pull request, #25086:
URL: https://github.com/apache/pulsar/pull/25086
### Motivation
It is not recommended to use sync operations in NIO threads, which would block other operations and degrade the overall performance of the system.
### Modifications
Refactor namespace anti affinity group sync operations to async in rest api, and add tests.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [x] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: https://github.com/oneby-wang/pulsar/pull/14
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #25086:
URL: https://github.com/apache/pulsar/pull/25086#discussion_r2626553501
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -1842,104 +1811,69 @@ protected void internalSetDelayedDelivery(DelayedDeliveryPolicies delayedDeliver
internalSetPolicies("delayed_delivery_policies", delayedDeliveryPolicies);
}
- protected void internalSetNamespaceAntiAffinityGroup(String antiAffinityGroup) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Setting anti-affinity group {} for {}", clientAppId(), antiAffinityGroup, namespaceName);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "antiAffinityGroup can't be empty");
- }
-
- try {
- getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, (lp)->
- lp.map(policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- antiAffinityGroup,
- policies.migrated))
- .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, antiAffinityGroup))
- );
- log.info("[{}] Successfully updated local-policies configuration: namespace={}, map={}", clientAppId(),
- namespaceName, antiAffinityGroup);
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("[{}] Failed to update local-policy configuration for namespace {}", clientAppId(), namespaceName,
- e);
- throw new RestException(e);
- }
- }
-
- protected String internalGetNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
-
- try {
- return getLocalPolicies()
- .getLocalPolicies(namespaceName)
- .orElseGet(() -> new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles())
- , null, null)).namespaceAntiAffinityGroup;
- } catch (Exception e) {
- log.error("[{}] Failed to get the antiAffinityGroup of namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(Status.NOT_FOUND, "Couldn't find namespace policies");
- }
- }
-
- protected void internalRemoveNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Deleting anti-affinity group for {}", clientAppId(), namespaceName);
-
- try {
- getLocalPolicies().setLocalPolicies(namespaceName, (policies)->
- new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- null,
- policies.migrated));
- log.info("[{}] Successfully removed anti-affinity group for a namespace={}", clientAppId(), namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to remove anti-affinity group for namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(e);
- }
+ protected CompletableFuture<Void> internalSetNamespaceAntiAffinityGroupAsync(String antiAffinityGroup) {
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()).thenCompose(
+ __ -> getDefaultBundleDataAsync().thenCompose(
+ defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
+ oldPolicies -> oldPolicies.map(policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup, antiAffinityGroup,
+ policies.migrated))
+ .orElseGet(() -> new LocalPolicies(defaultBundleData, null,
+ antiAffinityGroup)))))
+ .thenAccept(__ -> log.info(
+ "[{}] Successfully updated namespace anti-affinity group, namespace={}, anti-affinity"
+ + " group={}", clientAppId(), namespaceName, antiAffinityGroup));
+ }
+
+ protected CompletableFuture<String> internalGetNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> getLocalPolicies().getLocalPoliciesAsync(namespaceName)
+ .thenApply(policiesOpt -> policiesOpt.map(localPolicies -> localPolicies.namespaceAntiAffinityGroup)
+ .orElse(null)));
+ }
+
+ protected CompletableFuture<Void> internalRemoveNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+ .thenCompose(__ -> {
+ log.info("[{}] Removing anti-affinity group for namespace: {}", clientAppId(), namespaceName);
+ return getLocalPolicies().setLocalPoliciesAsync(namespaceName,
+ (policies) -> new LocalPolicies(policies.bundles, policies.bookieAffinityGroup, null,
+ policies.migrated));
+ })
+ .thenAccept(__ -> log.info("[{}] Successfully removed anti-affinity group for namespace: {}",
+ clientAppId(), namespaceName));
}
- protected List<String> internalGetAntiAffinityNamespaces(String cluster, String antiAffinityGroup,
- String tenant) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
+ protected CompletableFuture<List<String>> internalGetAntiAffinityNamespacesAsync(String cluster,
+ String antiAffinityGroup,
+ String tenant) {
checkNotNull(cluster, "Cluster should not be null");
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
checkNotNull(tenant, "Tenant should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
- log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup, cluster);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "anti-affinity group can't be empty.");
- }
- validateClusterExists(cluster);
-
- try {
- List<String> namespaces = tenantResources().getListOfNamespaces(tenant);
-
- return namespaces.stream().filter(ns -> {
- Optional<LocalPolicies> policies;
- try {
- policies = getLocalPolicies().getLocalPolicies(NamespaceName.get(ns));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- String storedAntiAffinityGroup = policies.orElseGet(() ->
- new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles()),
- null, null)).namespaceAntiAffinityGroup;
- return antiAffinityGroup.equalsIgnoreCase(storedAntiAffinityGroup);
- }).collect(Collectors.toList());
-
- } catch (Exception e) {
- log.warn("Failed to list of properties/namespace from global-zk", e);
- throw new RestException(e);
- }
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> validateClusterExistsAsync(cluster))
+ .thenCompose(__ -> {
+ log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup,
+ cluster);
+ return tenantResources().getListOfNamespacesAsync(tenant).thenCompose(namespaces -> {
+ List<CompletableFuture<String>> nsFutures = namespaces.stream()
+ .map(ns -> getLocalPolicies().getLocalPoliciesAsync(NamespaceName.get(ns))
+ .thenApply(policiesOpt -> policiesOpt.map(
+ localPolicies -> localPolicies.namespaceAntiAffinityGroup).orElse(null))
+ .thenApply(antiAffinityGroup::equalsIgnoreCase)
Review Comment:
Potential NullPointerException: when the namespace's anti-affinity group is null (line 1868), calling `antiAffinityGroup::equalsIgnoreCase` will fail. The method reference expects a String but receives null, which causes the comparison to fail. Replace this with a null-safe comparison such as using `Objects.equals` or adding an explicit null check before the comparison.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
Posted by "oneby-wang (via GitHub)" <gi...@apache.org>.
oneby-wang commented on code in PR #25086:
URL: https://github.com/apache/pulsar/pull/25086#discussion_r2626608923
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -1842,104 +1811,69 @@ protected void internalSetDelayedDelivery(DelayedDeliveryPolicies delayedDeliver
internalSetPolicies("delayed_delivery_policies", delayedDeliveryPolicies);
}
- protected void internalSetNamespaceAntiAffinityGroup(String antiAffinityGroup) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Setting anti-affinity group {} for {}", clientAppId(), antiAffinityGroup, namespaceName);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "antiAffinityGroup can't be empty");
- }
-
- try {
- getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, (lp)->
- lp.map(policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- antiAffinityGroup,
- policies.migrated))
- .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, antiAffinityGroup))
- );
- log.info("[{}] Successfully updated local-policies configuration: namespace={}, map={}", clientAppId(),
- namespaceName, antiAffinityGroup);
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("[{}] Failed to update local-policy configuration for namespace {}", clientAppId(), namespaceName,
- e);
- throw new RestException(e);
- }
- }
-
- protected String internalGetNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
-
- try {
- return getLocalPolicies()
- .getLocalPolicies(namespaceName)
- .orElseGet(() -> new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles())
- , null, null)).namespaceAntiAffinityGroup;
- } catch (Exception e) {
- log.error("[{}] Failed to get the antiAffinityGroup of namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(Status.NOT_FOUND, "Couldn't find namespace policies");
- }
- }
-
- protected void internalRemoveNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Deleting anti-affinity group for {}", clientAppId(), namespaceName);
-
- try {
- getLocalPolicies().setLocalPolicies(namespaceName, (policies)->
- new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- null,
- policies.migrated));
- log.info("[{}] Successfully removed anti-affinity group for a namespace={}", clientAppId(), namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to remove anti-affinity group for namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(e);
- }
+ protected CompletableFuture<Void> internalSetNamespaceAntiAffinityGroupAsync(String antiAffinityGroup) {
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()).thenCompose(
+ __ -> getDefaultBundleDataAsync().thenCompose(
+ defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
+ oldPolicies -> oldPolicies.map(policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup, antiAffinityGroup,
+ policies.migrated))
+ .orElseGet(() -> new LocalPolicies(defaultBundleData, null,
+ antiAffinityGroup)))))
+ .thenAccept(__ -> log.info(
+ "[{}] Successfully updated namespace anti-affinity group, namespace={}, anti-affinity"
+ + " group={}", clientAppId(), namespaceName, antiAffinityGroup));
+ }
+
+ protected CompletableFuture<String> internalGetNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> getLocalPolicies().getLocalPoliciesAsync(namespaceName)
+ .thenApply(policiesOpt -> policiesOpt.map(localPolicies -> localPolicies.namespaceAntiAffinityGroup)
+ .orElse(null)));
+ }
+
+ protected CompletableFuture<Void> internalRemoveNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+ .thenCompose(__ -> {
+ log.info("[{}] Removing anti-affinity group for namespace: {}", clientAppId(), namespaceName);
+ return getLocalPolicies().setLocalPoliciesAsync(namespaceName,
+ (policies) -> new LocalPolicies(policies.bundles, policies.bookieAffinityGroup, null,
+ policies.migrated));
+ })
+ .thenAccept(__ -> log.info("[{}] Successfully removed anti-affinity group for namespace: {}",
+ clientAppId(), namespaceName));
}
- protected List<String> internalGetAntiAffinityNamespaces(String cluster, String antiAffinityGroup,
- String tenant) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
+ protected CompletableFuture<List<String>> internalGetAntiAffinityNamespacesAsync(String cluster,
+ String antiAffinityGroup,
+ String tenant) {
checkNotNull(cluster, "Cluster should not be null");
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
checkNotNull(tenant, "Tenant should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
- log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup, cluster);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "anti-affinity group can't be empty.");
- }
- validateClusterExists(cluster);
-
- try {
- List<String> namespaces = tenantResources().getListOfNamespaces(tenant);
-
- return namespaces.stream().filter(ns -> {
- Optional<LocalPolicies> policies;
- try {
- policies = getLocalPolicies().getLocalPolicies(NamespaceName.get(ns));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- String storedAntiAffinityGroup = policies.orElseGet(() ->
- new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles()),
- null, null)).namespaceAntiAffinityGroup;
- return antiAffinityGroup.equalsIgnoreCase(storedAntiAffinityGroup);
- }).collect(Collectors.toList());
-
- } catch (Exception e) {
- log.warn("Failed to list of properties/namespace from global-zk", e);
- throw new RestException(e);
- }
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> validateClusterExistsAsync(cluster))
+ .thenCompose(__ -> {
+ log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup,
+ cluster);
+ return tenantResources().getListOfNamespacesAsync(tenant).thenCompose(namespaces -> {
+ List<CompletableFuture<String>> nsFutures = namespaces.stream()
+ .map(ns -> getLocalPolicies().getLocalPoliciesAsync(NamespaceName.get(ns))
+ .thenApply(policiesOpt -> policiesOpt.map(
+ localPolicies -> localPolicies.namespaceAntiAffinityGroup).orElse(null))
+ .thenApply(antiAffinityGroup::equalsIgnoreCase)
Review Comment:
String#equalsIgnoreCase() source code:
```java
public boolean equalsIgnoreCase(String anotherString) {
return (this == anotherString) ? true
: (anotherString != null)
&& (anotherString.length() == length())
&& regionMatches(true, 0, anotherString, 0, length());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #25086:
URL: https://github.com/apache/pulsar/pull/25086#issuecomment-3663564707
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/25086?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `73.78641%` with `27 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.43%. Comparing base ([`c8dbc3c`](https://app.codecov.io/gh/apache/pulsar/commit/c8dbc3c9c2276d1324adc53f33a2f409238d121f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`62046b6`](https://app.codecov.io/gh/apache/pulsar/commit/62046b6e471fec6d0ab41511201c5b75e7adf9b6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 10 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/25086?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../org/apache/pulsar/broker/admin/v1/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv1%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92MS9OYW1lc3BhY2VzLmphdmE=) | 42.85% | [16 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../org/apache/pulsar/broker/admin/v2/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv2%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9OYW1lc3BhY2VzLmphdmE=) | 71.42% | [8 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../org/apache/pulsar/broker/admin/AdminResource.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2FAdminResource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9BZG1pblJlc291cmNlLmphdmE=) | 50.00% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #25086 +/- ##
============================================
- Coverage 74.54% 74.43% -0.11%
+ Complexity 34256 2346 -31910
============================================
Files 1921 1899 -22
Lines 150526 149652 -874
Branches 17495 17393 -102
============================================
- Hits 112215 111399 -816
+ Misses 29452 29374 -78
- Partials 8859 8879 +20
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/25086/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/25086/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.29% <0.00%> (-0.71%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/25086/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.02% <0.00%> (+0.02%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/25086/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.98% <73.78%> (-0.10%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/25086?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ulsar/broker/resources/LocalPoliciesResources.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fresources%2FLocalPoliciesResources.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9icm9rZXIvcmVzb3VyY2VzL0xvY2FsUG9saWNpZXNSZXNvdXJjZXMuamF2YQ==) | `75.86% <100.00%> (-0.81%)` | :arrow_down: |
| [...pache/pulsar/broker/admin/impl/NamespacesBase.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fimpl%2FNamespacesBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL05hbWVzcGFjZXNCYXNlLmphdmE=) | `77.59% <100.00%> (+1.69%)` | :arrow_up: |
| [.../org/apache/pulsar/broker/admin/AdminResource.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2FAdminResource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9BZG1pblJlc291cmNlLmphdmE=) | `75.61% <50.00%> (+0.04%)` | :arrow_up: |
| [.../org/apache/pulsar/broker/admin/v2/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv2%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92Mi9OYW1lc3BhY2VzLmphdmE=) | `90.59% <71.42%> (-0.43%)` | :arrow_down: |
| [.../org/apache/pulsar/broker/admin/v1/Namespaces.java](https://app.codecov.io/gh/apache/pulsar/pull/25086?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fv1%2FNamespaces.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi92MS9OYW1lc3BhY2VzLmphdmE=) | `52.45% <42.85%> (-0.57%)` | :arrow_down: |
... and [116 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/25086/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #25086:
URL: https://github.com/apache/pulsar/pull/25086
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
Posted by "oneby-wang (via GitHub)" <gi...@apache.org>.
oneby-wang commented on code in PR #25086:
URL: https://github.com/apache/pulsar/pull/25086#discussion_r2626608923
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -1842,104 +1811,69 @@ protected void internalSetDelayedDelivery(DelayedDeliveryPolicies delayedDeliver
internalSetPolicies("delayed_delivery_policies", delayedDeliveryPolicies);
}
- protected void internalSetNamespaceAntiAffinityGroup(String antiAffinityGroup) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Setting anti-affinity group {} for {}", clientAppId(), antiAffinityGroup, namespaceName);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "antiAffinityGroup can't be empty");
- }
-
- try {
- getLocalPolicies().setLocalPoliciesWithCreate(namespaceName, (lp)->
- lp.map(policies -> new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- antiAffinityGroup,
- policies.migrated))
- .orElseGet(() -> new LocalPolicies(getDefaultBundleData(), null, antiAffinityGroup))
- );
- log.info("[{}] Successfully updated local-policies configuration: namespace={}, map={}", clientAppId(),
- namespaceName, antiAffinityGroup);
- } catch (RestException re) {
- throw re;
- } catch (Exception e) {
- log.error("[{}] Failed to update local-policy configuration for namespace {}", clientAppId(), namespaceName,
- e);
- throw new RestException(e);
- }
- }
-
- protected String internalGetNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
-
- try {
- return getLocalPolicies()
- .getLocalPolicies(namespaceName)
- .orElseGet(() -> new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles())
- , null, null)).namespaceAntiAffinityGroup;
- } catch (Exception e) {
- log.error("[{}] Failed to get the antiAffinityGroup of namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(Status.NOT_FOUND, "Couldn't find namespace policies");
- }
- }
-
- protected void internalRemoveNamespaceAntiAffinityGroup() {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE);
- validatePoliciesReadOnlyAccess();
-
- log.info("[{}] Deleting anti-affinity group for {}", clientAppId(), namespaceName);
-
- try {
- getLocalPolicies().setLocalPolicies(namespaceName, (policies)->
- new LocalPolicies(policies.bundles,
- policies.bookieAffinityGroup,
- null,
- policies.migrated));
- log.info("[{}] Successfully removed anti-affinity group for a namespace={}", clientAppId(), namespaceName);
- } catch (Exception e) {
- log.error("[{}] Failed to remove anti-affinity group for namespace {}", clientAppId(), namespaceName, e);
- throw new RestException(e);
- }
+ protected CompletableFuture<Void> internalSetNamespaceAntiAffinityGroupAsync(String antiAffinityGroup) {
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync()).thenCompose(
+ __ -> getDefaultBundleDataAsync().thenCompose(
+ defaultBundleData -> getLocalPolicies().setLocalPoliciesWithCreateAsync(namespaceName,
+ oldPolicies -> oldPolicies.map(policies -> new LocalPolicies(policies.bundles,
+ policies.bookieAffinityGroup, antiAffinityGroup,
+ policies.migrated))
+ .orElseGet(() -> new LocalPolicies(defaultBundleData, null,
+ antiAffinityGroup)))))
+ .thenAccept(__ -> log.info(
+ "[{}] Successfully updated namespace anti-affinity group, namespace={}, anti-affinity"
+ + " group={}", clientAppId(), namespaceName, antiAffinityGroup));
+ }
+
+ protected CompletableFuture<String> internalGetNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> getLocalPolicies().getLocalPoliciesAsync(namespaceName)
+ .thenApply(policiesOpt -> policiesOpt.map(localPolicies -> localPolicies.namespaceAntiAffinityGroup)
+ .orElse(null)));
+ }
+
+ protected CompletableFuture<Void> internalRemoveNamespaceAntiAffinityGroupAsync() {
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.WRITE)
+ .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+ .thenCompose(__ -> {
+ log.info("[{}] Removing anti-affinity group for namespace: {}", clientAppId(), namespaceName);
+ return getLocalPolicies().setLocalPoliciesAsync(namespaceName,
+ (policies) -> new LocalPolicies(policies.bundles, policies.bookieAffinityGroup, null,
+ policies.migrated));
+ })
+ .thenAccept(__ -> log.info("[{}] Successfully removed anti-affinity group for namespace: {}",
+ clientAppId(), namespaceName));
}
- protected List<String> internalGetAntiAffinityNamespaces(String cluster, String antiAffinityGroup,
- String tenant) {
- validateNamespacePolicyOperation(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ);
+ protected CompletableFuture<List<String>> internalGetAntiAffinityNamespacesAsync(String cluster,
+ String antiAffinityGroup,
+ String tenant) {
checkNotNull(cluster, "Cluster should not be null");
- checkNotNull(antiAffinityGroup, "AntiAffinityGroup should not be null");
+ checkNotNull(antiAffinityGroup, "Anti-affinity group should not be null");
checkNotNull(tenant, "Tenant should not be null");
+ checkNotBlank(antiAffinityGroup, "Anti-affinity group can't be empty");
- log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup, cluster);
-
- if (isBlank(antiAffinityGroup)) {
- throw new RestException(Status.PRECONDITION_FAILED, "anti-affinity group can't be empty.");
- }
- validateClusterExists(cluster);
-
- try {
- List<String> namespaces = tenantResources().getListOfNamespaces(tenant);
-
- return namespaces.stream().filter(ns -> {
- Optional<LocalPolicies> policies;
- try {
- policies = getLocalPolicies().getLocalPolicies(NamespaceName.get(ns));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- String storedAntiAffinityGroup = policies.orElseGet(() ->
- new LocalPolicies(getBundles(config().getDefaultNumberOfNamespaceBundles()),
- null, null)).namespaceAntiAffinityGroup;
- return antiAffinityGroup.equalsIgnoreCase(storedAntiAffinityGroup);
- }).collect(Collectors.toList());
-
- } catch (Exception e) {
- log.warn("Failed to list of properties/namespace from global-zk", e);
- throw new RestException(e);
- }
+ return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ANTI_AFFINITY, PolicyOperation.READ)
+ .thenCompose(__ -> validateClusterExistsAsync(cluster))
+ .thenCompose(__ -> {
+ log.info("[{}]-{} Finding namespaces for {} in {}", clientAppId(), tenant, antiAffinityGroup,
+ cluster);
+ return tenantResources().getListOfNamespacesAsync(tenant).thenCompose(namespaces -> {
+ List<CompletableFuture<String>> nsFutures = namespaces.stream()
+ .map(ns -> getLocalPolicies().getLocalPoliciesAsync(NamespaceName.get(ns))
+ .thenApply(policiesOpt -> policiesOpt.map(
+ localPolicies -> localPolicies.namespaceAntiAffinityGroup).orElse(null))
+ .thenApply(antiAffinityGroup::equalsIgnoreCase)
Review Comment:
`String#equalsIgnoreCase()` source code:
```java
public boolean equalsIgnoreCase(String anotherString) {
return (this == anotherString) ? true
: (anotherString != null)
&& (anotherString.length() == length())
&& regionMatches(true, 0, anotherString, 0, length());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][admin] Refactor namespace anti affinity group sync operations to async in rest api [pulsar]
Posted by "oneby-wang (via GitHub)" <gi...@apache.org>.
oneby-wang commented on PR #25086:
URL: https://github.com/apache/pulsar/pull/25086#issuecomment-3663493456
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org