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/08/28 12:29:26 UTC
(pulsar) branch branch-4.0 updated: [fix][test] Fix flaky IsolatedBookieEnsemblePlacementPolicyTest by awaiting the initial rack config load (#26429)
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
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 3c6ea49b8999 [fix][test] Fix flaky IsolatedBookieEnsemblePlacementPolicyTest by awaiting the initial rack config load (#26429)
3c6ea49b8999 is described below
commit 3c6ea49b899998bd921920e38e785a8d94ac1097
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Aug 28 13:34:18 2026 +0300
[fix][test] Fix flaky IsolatedBookieEnsemblePlacementPolicyTest by awaiting the initial rack config load (#26429)
(cherry picked from commit 5180dba3327d2cfe4668dc69ac4418efe3a24b0f)
---
.../IsolatedBookieEnsemblePlacementPolicy.java | 32 +++++-
.../IsolatedBookieEnsemblePlacementPolicyTest.java | 127 ++++++++++++---------
2 files changed, 98 insertions(+), 61 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 f2c1b5e73619..00da83db4816 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
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException;
@@ -67,6 +68,22 @@ public class IsolatedBookieEnsemblePlacementPolicy extends RackawareEnsemblePlac
private volatile BookiesRackConfiguration cachedRackConfiguration = null;
+ /**
+ * Completes once the rack configuration load started by
+ * {@link #initialize(ClientConfiguration, Optional, HashedWheelTimer, FeatureProvider, StatsLogger,
+ * BookieAddressResolver)} has been applied to {@link #cachedRackConfiguration}, and completes exceptionally
+ * when that load failed. Until it completes no isolation is applied at all, so tests must wait for this future
+ * before asserting on placement decisions.
+ *
+ * <p>It deliberately excludes the {@code exceptionally} stage that keeps initialization going on a failed
+ * load: a test awaiting this future must see the real failure rather than proceed against a still-null
+ * {@link #cachedRackConfiguration}, which looks exactly like the race this future exists to close.
+ */
+ @Getter
+ @VisibleForTesting
+ private volatile CompletableFuture<Void> initialRackConfigurationLoadFuture =
+ CompletableFuture.completedFuture(null);
+
public IsolatedBookieEnsemblePlacementPolicy() {
super();
}
@@ -91,12 +108,15 @@ public class IsolatedBookieEnsemblePlacementPolicy extends RackawareEnsemblePlac
}
// Only add the bookieMappingCache if we have defined an isolation group
bookieMappingCache = store.getMetadataCache(BookiesRackConfiguration.class);
- bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).thenAccept(opt -> opt.ifPresent(
- bookiesRackConfiguration -> cachedRackConfiguration = bookiesRackConfiguration))
- .exceptionally(e -> {
- log.warn("Failed to load bookies rack configuration while initialize the PlacementPolicy.");
- return null;
- });
+ CompletableFuture<Void> rackConfigurationLoad = bookieMappingCache
+ .get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).thenAccept(opt -> opt.ifPresent(
+ bookiesRackConfiguration -> cachedRackConfiguration = bookiesRackConfiguration));
+ // Initialization continues when the load fails; isolation is simply not applied until a later refresh.
+ rackConfigurationLoad.exceptionally(e -> {
+ log.warn("Failed to load bookies rack configuration while initialize the PlacementPolicy.", e);
+ return null;
+ });
+ initialRackConfigurationLoadFuture = rackConfigurationLoad;
}
if (conf.getProperty(SECONDARY_ISOLATION_BOOKIE_GROUPS) != null) {
String secondaryIsolationGroupsString = ConfigurationStringUtil
diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java
index d4785d6ca83d..b54840e71f6b 100644
--- a/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java
+++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/IsolatedBookieEnsemblePlacementPolicyTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.pulsar.bookie.rackawareness;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -41,6 +43,8 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException;
import org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy;
import org.apache.bookkeeper.conf.ClientConfiguration;
@@ -56,6 +60,7 @@ import org.apache.pulsar.common.policies.data.EnsemblePlacementPolicyConfig;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl;
@@ -73,6 +78,7 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
private static final String BOOKIE3 = "127.0.0.3:3181";
private static final String BOOKIE4 = "127.0.0.4:3181";
private static final String BOOKIE5 = "127.0.0.5:3181";
+ private static final int RACK_CONFIGURATION_LOAD_TIMEOUT_SECONDS = 30;
private MetadataStore store;
private final ObjectMapper jsonMapper = ObjectMapperFactory.create();
@@ -116,12 +122,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroups);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
List<BookieId> ensemble = isolationPolicy.newEnsemble(2, 2, 2,
@@ -190,12 +194,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
.thenReturn(waitingCompleteFuture).thenReturn(waitingCompleteFuture)
.thenReturn(emptyFuture).thenReturn(emptyFuture);
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroups);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
MutablePair<Set<String>, Set<String>> groups = new MutablePair<>();
@@ -255,12 +257,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroups);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
List<BookieId> ensemble = isolationPolicy.newEnsemble(3, 3, 2,
@@ -329,12 +329,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
@Test
public void testNoBookieInfo() throws Exception {
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroups);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
isolationPolicy.newEnsemble(4, 4, 4, Collections.emptyMap(), new HashSet<>());
@@ -378,12 +376,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroups);
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
List<BookieId> ensemble = isolationPolicy.newEnsemble(2, 2, 2,
@@ -425,11 +421,9 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
Awaitility.await().until(() -> store.exists(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).join());
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
isolationPolicy.newEnsemble(4, 4, 4, Collections.emptyMap(), new HashSet<>());
@@ -503,12 +497,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolatedGroup);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
List<BookieId> ensemble = isolationPolicy
@@ -545,14 +537,12 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolatedGroup);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS,
secondaryIsolatedGroup);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
List<BookieId> ensemble = isolationPolicy
@@ -585,22 +575,14 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolatedGroup);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS,
secondaryIsolatedGroup);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
- // Wait for the async cache load triggered by initialize() to complete; otherwise the
- // first newEnsemble call can race with an empty cachedRackConfiguration and skip isolation.
- Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
- assertNotNull(isolationPolicy.getBookieMappingCache()
- .getIfCached(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH)));
-
try {
isolationPolicy
.newEnsemble(3, 3, 2, Collections.emptyMap(), new HashSet<>()).getResult();
@@ -651,12 +633,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
customMetadata.put(EnsemblePlacementPolicyConfig.ENSEMBLE_PLACEMENT_POLICY_CONFIG, policyConfig.encode());
// do the test logic
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, primaryGroupName);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
// we assume we have an ensemble list which is consist with bookie1 and bookie3, and bookie3 is broken.
@@ -698,14 +678,12 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, defaultIsolatedGroup);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS,
defaultSecondaryIsolatedGroup);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
Map<String, Object> placementPolicyProperties = new HashMap<>();
@@ -750,23 +728,14 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroup1);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS,
isolationGroup2);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
- // Wait for the async cache load triggered by initialize() to complete; otherwise
- // getExcludedBookiesWithIsolationGroups returns an empty set when cachedRackConfiguration
- // is still null. Same pattern used in testBookieInfoChange (#25473).
- Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() ->
- assertNotNull(isolationPolicy.getBookieMappingCache()
- .getIfCached(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH)));
-
/* Test common cases */
MutablePair<Set<String>, Set<String>> groups = new MutablePair<>();
groups.setLeft(Sets.newHashSet(isolationGroup1));
@@ -881,12 +850,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, "group1");
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
// Use a policy class that does NOT match IsolatedBookieEnsemblePlacementPolicy.
@@ -937,12 +904,10 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
store.put(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH, jsonMapper.writeValueAsBytes(bookieMapping),
Optional.empty()).join();
- IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
ClientConfiguration bkClientConf = new ClientConfiguration();
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, defaultGroup);
- isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
- NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = createIsolationPolicy(bkClientConf);
isolationPolicy.onClusterChanged(writableBookies, readOnlyBookies);
// --- unit-level: getIsolationGroup should parse properties, not fall back to defaults ---
@@ -986,6 +951,58 @@ public class IsolatedBookieEnsemblePlacementPolicyTest {
"default ensemble should come from " + defaultGroup + ", got " + defaultEnsemble);
}
+ /**
+ * A failed initial rack configuration load leaves {@code cachedRackConfiguration} null, which is
+ * indistinguishable from the load simply not having completed yet: no isolation is applied either way. The
+ * exposed future therefore reports that failure rather than completing normally, while initialization itself
+ * keeps its log-and-continue behaviour.
+ */
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testInitialRackConfigurationLoadFailureIsReportedWithoutFailingInitialize() throws Exception {
+ // teardown() only closes whatever `store` points at, so close the real one before replacing it.
+ store.close();
+ store = mock(MetadataStoreExtended.class);
+ MetadataCacheImpl<BookiesRackConfiguration> cache = mock(MetadataCacheImpl.class);
+ doReturn(cache).when(store).getMetadataCache(BookiesRackConfiguration.class);
+ MetadataStoreException loadFailure = new MetadataStoreException("simulated rack configuration load failure");
+ when(cache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH))
+ .thenReturn(CompletableFuture.failedFuture(loadFailure));
+
+ ClientConfiguration bkClientConf = new ClientConfiguration();
+ bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE, store);
+ bkClientConf.setProperty(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, isolationGroups);
+
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
+ // A failed load must not break initialization; that is the pre-existing production behaviour.
+ isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
+ NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+
+ assertThatThrownBy(() -> isolationPolicy.getInitialRackConfigurationLoadFuture()
+ .get(RACK_CONFIGURATION_LOAD_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .cause().isSameAs(loadFailure);
+ }
+
+ /**
+ * Creates and initializes the policy under test, and waits until the rack configuration load started by
+ * {@code initialize} has been applied. That load is asynchronous, and until it completes
+ * {@code getExcludedBookiesWithIsolationGroups} finds a null {@code cachedRackConfiguration} and silently
+ * applies no isolation at all, so any placement assertion made before it completes is racy.
+ *
+ * <p>The wait is bounded and propagates a failed load, so that a load which failed or never completed fails
+ * the test with its own cause instead of surfacing later as an unexplained placement assertion failure.
+ */
+ private IsolatedBookieEnsemblePlacementPolicy createIsolationPolicy(ClientConfiguration bkClientConf)
+ throws Exception {
+ IsolatedBookieEnsemblePlacementPolicy isolationPolicy = new IsolatedBookieEnsemblePlacementPolicy();
+ isolationPolicy.initialize(bkClientConf, Optional.empty(), timer, SettableFeatureProvider.DISABLE_ALL,
+ NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
+ isolationPolicy.getInitialRackConfigurationLoadFuture().get(RACK_CONFIGURATION_LOAD_TIMEOUT_SECONDS,
+ TimeUnit.SECONDS);
+ return isolationPolicy;
+ }
+
// The policy gets the bookie info asynchronously before each query or update, when putting the bookie info into
// the metadata store, the cache needs some time to receive the notification and update accordingly.
private void updateBookieInfo(IsolatedBookieEnsemblePlacementPolicy isolationPolicy, byte[] bookieInfo) {