You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ba...@apache.org on 2023/10/26 07:07:34 UTC
[pulsar] branch branch-2.11 updated: [fix] [broker] [branch-2.11] Fix isolated group not work problem (#21415)
This is an automated email from the ASF dual-hosted git repository.
baodi pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new 42e8e7c6e4a [fix] [broker] [branch-2.11] Fix isolated group not work problem (#21415)
42e8e7c6e4a is described below
commit 42e8e7c6e4ad4f1613dd5f6f97863eaa5134af2e
Author: Yan Zhao <ho...@apache.org>
AuthorDate: Thu Oct 26 15:07:29 2023 +0800
[fix] [broker] [branch-2.11] Fix isolated group not work problem (#21415)
---
.../IsolatedBookieEnsemblePlacementPolicy.java | 22 ++++++++++++----------
1 file changed, 12 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicy.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicy.java
index 9e6affbb7ef..9062c25a21c 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicy.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicy.java
@@ -27,7 +27,8 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException;
import org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy;
@@ -47,6 +48,8 @@ import org.apache.pulsar.common.policies.data.BookiesRackConfiguration;
import org.apache.pulsar.common.policies.data.EnsemblePlacementPolicyConfig;
import org.apache.pulsar.metadata.api.MetadataCache;
import org.apache.pulsar.metadata.api.MetadataStore;
+import org.apache.pulsar.zookeeper.ZkBookieRackAffinityMapping;
+import org.apache.zookeeper.KeeperException;
@Slf4j
public class IsolatedBookieEnsemblePlacementPolicy extends RackawareEnsemblePlacementPolicy {
@@ -193,17 +196,14 @@ public class IsolatedBookieEnsemblePlacementPolicy extends RackawareEnsemblePlac
}
try {
if (bookieMappingCache != null) {
- CompletableFuture<Optional<BookiesRackConfiguration>> future =
- bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH);
-
- Optional<BookiesRackConfiguration> optRes = (future.isDone() && !future.isCompletedExceptionally())
- ? future.join() : Optional.empty();
-
- if (!optRes.isPresent()) {
- return blacklistedBookies;
+ Optional<BookiesRackConfiguration> optional =
+ bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH)
+ .get(30, TimeUnit.SECONDS);
+ if (!optional.isPresent()) {
+ throw new KeeperException.NoNodeException(ZkBookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH);
}
- BookiesRackConfiguration allGroupsBookieMapping = optRes.get();
+ BookiesRackConfiguration allGroupsBookieMapping = optional.get();
Set<String> allBookies = allGroupsBookieMapping.keySet();
int totalAvailableBookiesInPrimaryGroup = 0;
Set<String> primaryIsolationGroup = Collections.emptySet();
@@ -272,6 +272,8 @@ public class IsolatedBookieEnsemblePlacementPolicy extends RackawareEnsemblePlac
}
}
}
+ } catch (TimeoutException e) {
+ log.warn("Getting bookie isolation info from metadata store timeout.");
} catch (Exception e) {
log.warn("Error getting bookie isolation info from metadata store: {}", e.getMessage());
}