This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 42fe150 Enhance EnsemblePlacementPolicy and DNSResolverDecorator to
log relevant metrics.
42fe150 is described below
commit 42fe1506afe04b8068b897c530c7ca7188da503e
Author: Charan Reddy Guttapalem <[email protected]>
AuthorDate: Sat Jan 12 18:04:35 2019 -0800
Enhance EnsemblePlacementPolicy and DNSResolverDecorator to log relevant
metrics.
Descriptions of the changes in this PR:
Make changes to EnsemblePlacementPolicy so that it would return boolean
value indicating
if the return value of newEnsemble and replaceBookie are strictly adhering
to corresponding
PlacementPolicy or it fell back to random.
Similarly DNSResolverDecorator should log a metric when it was unable to
resolve rack info and
it is using default rack.
Reviewers: Samuel Just <[email protected]>, Sijie Guo
<[email protected]>, Venkateswararao Jujjuri (JV) <None>
This closes #1883 from reddycharan/enhanceplacementpolicy
---
.../bookkeeper/bookie/BookKeeperServerStats.java | 2 +
.../bookie/LocalBookieEnsemblePlacementPolicy.java | 19 ++-
.../apache/bookkeeper/client/BookKeeperAdmin.java | 16 +-
.../bookkeeper/client/BookieWatcherImpl.java | 71 +++++++--
.../client/DefaultEnsemblePlacementPolicy.java | 29 +++-
.../bookkeeper/client/EnsemblePlacementPolicy.java | 50 ++++--
.../ITopologyAwareEnsemblePlacementPolicy.java | 3 +-
.../client/RackawareEnsemblePlacementPolicy.java | 15 +-
.../RackawareEnsemblePlacementPolicyImpl.java | 100 ++++++++++--
.../client/RegionAwareEnsemblePlacementPolicy.java | 42 ++++-
.../apache/bookkeeper/net/ScriptBasedMapping.java | 8 +-
.../client/GenericEnsemblePlacementPolicyTest.java | 7 +-
.../TestRackawareEnsemblePlacementPolicy.java | 177 +++++++++++++++------
...ackawareEnsemblePlacementPolicyUsingScript.java | 45 ++++--
.../TestRackawarePolicyNotificationUpdates.java | 9 +-
.../TestRegionAwareEnsemblePlacementPolicy.java | 108 ++++++++-----
16 files changed, 515 insertions(+), 186 deletions(-)
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java
index cdafd15..b58514b 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java
@@ -69,6 +69,8 @@ public interface BookKeeperServerStats {
String WATCHER_SCOPE = "bookie_watcher";
String REPLACE_BOOKIE_TIME = "REPLACE_BOOKIE_TIME";
String NEW_ENSEMBLE_TIME = "NEW_ENSEMBLE_TIME";
+ String FAILED_TO_RESOLVE_NETWORK_LOCATION_COUNTER =
"FAILED_TO_RESOLVE_NETWORK_LOCATION_COUNTER";
+ String ENSEMBLE_NOT_ADHERING_TO_PLACEMENT_POLICY_COUNTER =
"ENSEMBLE_NOT_ADHERING_TO_PLACEMENT_POLICY_COUNTER";
// Bookie Operations
String BOOKIE_ADD_ENTRY = "BOOKIE_ADD_ENTRY";
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LocalBookieEnsemblePlacementPolicy.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LocalBookieEnsemblePlacementPolicy.java
index 46978ea..7b7cc46 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LocalBookieEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LocalBookieEnsemblePlacementPolicy.java
@@ -36,6 +36,7 @@ import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -80,9 +81,9 @@ public class LocalBookieEnsemblePlacementPolicy implements
EnsemblePlacementPoli
}
@Override
- public BookieSocketAddress replaceBookie(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- java.util.Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
currentEnsemble,
- BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
+ public Pair<BookieSocketAddress, Boolean> replaceBookie(int ensembleSize,
int writeQuorumSize, int ackQuorumSize,
+ java.util.Map<String, byte[]> customMetadata,
List<BookieSocketAddress> currentEnsemble,
+ BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
throws BKNotEnoughBookiesException {
throw new BKNotEnoughBookiesException();
}
@@ -109,18 +110,24 @@ public class LocalBookieEnsemblePlacementPolicy
implements EnsemblePlacementPoli
}
@Override
- public List<BookieSocketAddress> newEnsemble(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- java.util.Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
excludeBookies)
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize, int writeQuorumSize,
+ int ackQuorumSize, java.util.Map<String, byte[]> customMetadata,
Set<BookieSocketAddress> excludeBookies)
throws BKNotEnoughBookiesException {
if (ensembleSize > 1) {
throw new IllegalArgumentException("Local ensemble policy can only
return 1 bookie");
}
- return Lists.newArrayList(bookieAddress);
+ return Pair.of(Lists.newArrayList(bookieAddress), true);
}
@Override
public void updateBookieInfo(Map<BookieSocketAddress, BookieInfo>
bookieToFreeSpaceMap) {
return;
}
+
+ @Override
+ public boolean
isEnsembleAdheringToPlacementPolicy(List<BookieSocketAddress> ensembleList, int
writeQuorumSize,
+ int ackQuorumSize) {
+ return true;
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
index 3f837db..7b49292 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java
@@ -35,7 +35,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -82,6 +81,7 @@ import
org.apache.bookkeeper.replication.ReplicationException.UnavailableExcepti
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.IOUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
@@ -994,15 +994,25 @@ public class BookKeeperAdmin implements AutoCloseable {
// allocate bookies
for (Integer bookieIndex : bookieIndexesToRereplicate) {
BookieSocketAddress oldBookie = ensemble.get(bookieIndex);
- BookieSocketAddress newBookie =
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
bkc.getPlacementPolicy().replaceBookie(
lh.getLedgerMetadata().getEnsembleSize(),
lh.getLedgerMetadata().getWriteQuorumSize(),
lh.getLedgerMetadata().getAckQuorumSize(),
lh.getLedgerMetadata().getCustomMetadata(),
- new HashSet<>(ensemble),
+ ensemble,
oldBookie,
bookiesToExclude);
+ BookieSocketAddress newBookie = replaceBookieResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
+ if (!isEnsembleAdheringToPlacementPolicy) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "replaceBookie for bookie: {} in ensemble: {} "
+ + "is not adhering to placement policy and
chose {}",
+ oldBookie, ensemble, newBookie);
+ }
+ }
targetBookieAddresses.put(bookieIndex, newBookie);
bookiesToExclude.add(newBookie);
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcherImpl.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcherImpl.java
index 8b14b77..74c1df9 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcherImpl.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieWatcherImpl.java
@@ -17,6 +17,7 @@
*/
package org.apache.bookkeeper.client;
+import static
org.apache.bookkeeper.bookie.BookKeeperServerStats.ENSEMBLE_NOT_ADHERING_TO_PLACEMENT_POLICY_COUNTER;
import static
org.apache.bookkeeper.bookie.BookKeeperServerStats.NEW_ENSEMBLE_TIME;
import static
org.apache.bookkeeper.bookie.BookKeeperServerStats.REPLACE_BOOKIE_TIME;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.WATCHER_SCOPE;
@@ -35,6 +36,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.bookkeeper.bookie.BookKeeperServerStats;
import org.apache.bookkeeper.client.BKException.BKInterruptedException;
import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException;
import org.apache.bookkeeper.client.BKException.MetaStoreException;
@@ -43,9 +46,11 @@ import org.apache.bookkeeper.common.util.MathUtils;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.discover.RegistrationClient;
import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.annotations.StatsDoc;
+import org.apache.commons.lang3.tuple.Pair;
/**
* This class is responsible for maintaining a consistent view of what bookies
@@ -88,6 +93,12 @@ class BookieWatcherImpl implements BookieWatcher {
help = "operation stats of replacing bookie in an ensemble"
)
private final OpStatsLogger replaceBookieTimer;
+ @StatsDoc(
+ name = ENSEMBLE_NOT_ADHERING_TO_PLACEMENT_POLICY_COUNTER,
+ help = "total number of newEnsemble/replaceBookie operations
failed to adhere"
+ + " EnsemblePlacementPolicy"
+ )
+ private final Counter ensembleNotAdheringToPlacementPolicy;
// Bookies that will not be preferred to be chosen in a new ensemble
final Cache<BookieSocketAddress, Boolean> quarantinedBookies;
@@ -117,6 +128,8 @@ class BookieWatcherImpl implements BookieWatcher {
}).build();
this.newEnsembleTimer =
statsLogger.getOpStatsLogger(NEW_ENSEMBLE_TIME);
this.replaceBookieTimer =
statsLogger.getOpStatsLogger(REPLACE_BOOKIE_TIME);
+ this.ensembleNotAdheringToPlacementPolicy = statsLogger
+
.getCounter(BookKeeperServerStats.ENSEMBLE_NOT_ADHERING_TO_PLACEMENT_POLICY_COUNTER);
}
@Override
@@ -213,19 +226,34 @@ class BookieWatcherImpl implements BookieWatcher {
int ackQuorumSize, Map<String, byte[]> customMetadata)
throws BKNotEnoughBookiesException {
long startTime = MathUtils.nowInNano();
+ Pair<List<BookieSocketAddress>, Boolean> newEnsembleResponse;
List<BookieSocketAddress> socketAddresses;
+ boolean isEnsembleAdheringToPlacementPolicy = false;
try {
- socketAddresses = placementPolicy.newEnsemble(ensembleSize,
- writeQuorumSize, ackQuorumSize, customMetadata, new
HashSet<BookieSocketAddress>(
- quarantinedBookies.asMap().keySet()));
+ Set<BookieSocketAddress> quarantinedBookiesSet =
quarantinedBookies.asMap().keySet();
+ newEnsembleResponse = placementPolicy.newEnsemble(ensembleSize,
writeQuorumSize, ackQuorumSize,
+ customMetadata, new
HashSet<BookieSocketAddress>(quarantinedBookiesSet));
+ socketAddresses = newEnsembleResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy =
newEnsembleResponse.getRight();
+ if (!isEnsembleAdheringToPlacementPolicy) {
+ ensembleNotAdheringToPlacementPolicy.inc();
+ log.warn("New ensemble: {} is not adhering to Placement
Policy. quarantinedBookies: {}",
+ socketAddresses, quarantinedBookiesSet);
+ }
// we try to only get from the healthy bookies first
newEnsembleTimer.registerSuccessfulEvent(MathUtils.nowInNano() -
startTime, TimeUnit.NANOSECONDS);
} catch (BKNotEnoughBookiesException e) {
if (log.isDebugEnabled()) {
log.debug("Not enough healthy bookies available, using
quarantined bookies");
}
- socketAddresses = placementPolicy.newEnsemble(
+ newEnsembleResponse = placementPolicy.newEnsemble(
ensembleSize, writeQuorumSize, ackQuorumSize,
customMetadata, new HashSet<>());
+ socketAddresses = newEnsembleResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy =
newEnsembleResponse.getRight();
+ if (!isEnsembleAdheringToPlacementPolicy) {
+ ensembleNotAdheringToPlacementPolicy.inc();
+ log.warn("New ensemble: {} is not adhering to Placement
Policy", socketAddresses);
+ }
newEnsembleTimer.registerFailedEvent(MathUtils.nowInNano() -
startTime, TimeUnit.NANOSECONDS);
}
return socketAddresses;
@@ -239,22 +267,43 @@ class BookieWatcherImpl implements BookieWatcher {
throws BKNotEnoughBookiesException {
long startTime = MathUtils.nowInNano();
BookieSocketAddress addr = existingBookies.get(bookieIdx);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse;
BookieSocketAddress socketAddress;
+ boolean isEnsembleAdheringToPlacementPolicy = false;
try {
// we exclude the quarantined bookies also first
- Set<BookieSocketAddress> existingAndQuarantinedBookies = new
HashSet<BookieSocketAddress>(existingBookies);
-
existingAndQuarantinedBookies.addAll(quarantinedBookies.asMap().keySet());
- socketAddress = placementPolicy.replaceBookie(
+ Set<BookieSocketAddress> excludedBookiesAndQuarantinedBookies =
new HashSet<BookieSocketAddress>(
+ excludeBookies);
+ Set<BookieSocketAddress> quarantinedBookiesSet =
quarantinedBookies.asMap().keySet();
+ excludedBookiesAndQuarantinedBookies.addAll(quarantinedBookiesSet);
+ replaceBookieResponse = placementPolicy.replaceBookie(
ensembleSize, writeQuorumSize, ackQuorumSize,
customMetadata,
- existingAndQuarantinedBookies, addr, excludeBookies);
+ existingBookies, addr,
excludedBookiesAndQuarantinedBookies);
+ socketAddress = replaceBookieResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
+ if (!isEnsembleAdheringToPlacementPolicy) {
+ ensembleNotAdheringToPlacementPolicy.inc();
+ log.warn(
+ "replaceBookie for bookie: {} in ensemble: {} is not
adhering to placement policy and"
+ + " chose {}. excludedBookies {} and
quarantinedBookies {}",
+ addr, existingBookies, socketAddress, excludeBookies,
quarantinedBookiesSet);
+ }
replaceBookieTimer.registerSuccessfulEvent(MathUtils.nowInNano() -
startTime, TimeUnit.NANOSECONDS);
} catch (BKNotEnoughBookiesException e) {
if (log.isDebugEnabled()) {
log.debug("Not enough healthy bookies available, using
quarantined bookies");
}
- socketAddress = placementPolicy.replaceBookie(
- ensembleSize, writeQuorumSize, ackQuorumSize,
customMetadata,
- new HashSet<BookieSocketAddress>(existingBookies), addr,
excludeBookies);
+ replaceBookieResponse =
placementPolicy.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize,
+ customMetadata, existingBookies, addr, excludeBookies);
+ socketAddress = replaceBookieResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
+ if (!isEnsembleAdheringToPlacementPolicy) {
+ ensembleNotAdheringToPlacementPolicy.inc();
+ log.warn(
+ "replaceBookie for bookie: {} in ensemble: {} is not
adhering to placement policy and"
+ + " chose {}. excludedBookies {}",
+ addr, existingBookies, socketAddress, excludeBookies);
+ }
replaceBookieTimer.registerFailedEvent(MathUtils.nowInNano() -
startTime, TimeUnit.NANOSECONDS);
}
return socketAddress;
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/DefaultEnsemblePlacementPolicy.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/DefaultEnsemblePlacementPolicy.java
index 917d18f..dddbe1c 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/DefaultEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/DefaultEnsemblePlacementPolicy.java
@@ -39,6 +39,7 @@ import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,12 +65,12 @@ public class DefaultEnsemblePlacementPolicy implements
EnsemblePlacementPolicy {
}
@Override
- public ArrayList<BookieSocketAddress> newEnsemble(int ensembleSize, int
quorumSize, int ackQuorumSize,
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize, int quorumSize, int ackQuorumSize,
Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
excludeBookies)
throws BKNotEnoughBookiesException {
ArrayList<BookieSocketAddress> newBookies = new
ArrayList<BookieSocketAddress>(ensembleSize);
if (ensembleSize <= 0) {
- return newBookies;
+ return Pair.of(newBookies, false);
}
List<BookieSocketAddress> allBookies;
rwLock.readLock().lock();
@@ -95,7 +96,8 @@ public class DefaultEnsemblePlacementPolicy implements
EnsemblePlacementPolicy {
newBookies.add(b);
--ensembleSize;
if (ensembleSize == 0) {
- return newBookies;
+ return Pair.of(newBookies,
+
isEnsembleAdheringToPlacementPolicy(newBookies, quorumSize, ackQuorumSize));
}
}
} finally {
@@ -110,7 +112,8 @@ public class DefaultEnsemblePlacementPolicy implements
EnsemblePlacementPolicy {
newBookies.add(bookie);
--ensembleSize;
if (ensembleSize == 0) {
- return newBookies;
+ return Pair.of(newBookies,
+ isEnsembleAdheringToPlacementPolicy(newBookies,
quorumSize, ackQuorumSize));
}
}
}
@@ -118,13 +121,17 @@ public class DefaultEnsemblePlacementPolicy implements
EnsemblePlacementPolicy {
}
@Override
- public BookieSocketAddress replaceBookie(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
currentEnsemble,
+ public Pair<BookieSocketAddress, Boolean> replaceBookie(int ensembleSize,
int writeQuorumSize, int ackQuorumSize,
+ Map<String, byte[]> customMetadata, List<BookieSocketAddress>
currentEnsemble,
BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
throws BKNotEnoughBookiesException {
excludeBookies.addAll(currentEnsemble);
- ArrayList<BookieSocketAddress> addresses = newEnsemble(1, 1, 1,
customMetadata, excludeBookies);
- return addresses.get(0);
+ List<BookieSocketAddress> addresses = newEnsemble(1, 1, 1,
customMetadata, excludeBookies).getLeft();
+
+ BookieSocketAddress candidateAddr = addresses.get(0);
+ List<BookieSocketAddress> newEnsemble = new
ArrayList<BookieSocketAddress>(currentEnsemble);
+ newEnsemble.set(currentEnsemble.indexOf(bookieToReplace),
candidateAddr);
+ return Pair.of(candidateAddr,
isEnsembleAdheringToPlacementPolicy(newEnsemble, writeQuorumSize,
ackQuorumSize));
}
@Override
@@ -210,4 +217,10 @@ public class DefaultEnsemblePlacementPolicy implements
EnsemblePlacementPolicy {
public void uninitalize() {
// do nothing
}
+
+ @Override
+ public boolean
isEnsembleAdheringToPlacementPolicy(List<BookieSocketAddress> ensembleList, int
writeQuorumSize,
+ int ackQuorumSize) {
+ return true;
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/EnsemblePlacementPolicy.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/EnsemblePlacementPolicy.java
index 23932a3..00bac8e 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/EnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/EnsemblePlacementPolicy.java
@@ -35,6 +35,7 @@ import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.commons.lang3.tuple.Pair;
/**
* {@link EnsemblePlacementPolicy} encapsulates the algorithm that bookkeeper
client uses to select a number of bookies
@@ -263,12 +264,12 @@ public interface EnsemblePlacementPolicy {
* @throws BKNotEnoughBookiesException if not enough bookies available.
* @return the List<org.apache.bookkeeper.net.BookieSocketAddress>
*/
- List<BookieSocketAddress> newEnsemble(int ensembleSize,
- int writeQuorumSize,
- int ackQuorumSize,
- Map<String, byte[]>
customMetadata,
- Set<BookieSocketAddress>
excludeBookies)
- throws BKNotEnoughBookiesException;
+ Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int ensembleSize,
+ int writeQuorumSize,
+ int ackQuorumSize,
+ Map<String, byte[]>
customMetadata,
+
Set<BookieSocketAddress> excludeBookies)
+ throws BKNotEnoughBookiesException;
/**
* Choose a new bookie to replace <i>bookieToReplace</i>. If no bookie
available in the cluster,
@@ -287,14 +288,14 @@ public interface EnsemblePlacementPolicy {
* @throws BKNotEnoughBookiesException
* @return the org.apache.bookkeeper.net.BookieSocketAddress
*/
- BookieSocketAddress replaceBookie(int ensembleSize,
- int writeQuorumSize,
- int ackQuorumSize,
- Map<String, byte[]> customMetadata,
- Set<BookieSocketAddress> currentEnsemble,
- BookieSocketAddress bookieToReplace,
- Set<BookieSocketAddress> excludeBookies)
- throws BKNotEnoughBookiesException;
+ Pair<BookieSocketAddress, Boolean> replaceBookie(int ensembleSize,
+ int writeQuorumSize,
+ int ackQuorumSize,
+ Map<String, byte[]>
customMetadata,
+ List<BookieSocketAddress>
currentEnsemble,
+ BookieSocketAddress
bookieToReplace,
+ Set<BookieSocketAddress>
excludeBookies)
+ throws BKNotEnoughBookiesException;
/**
* Register a bookie as slow so that it is tried after available and
read-only bookies.
@@ -386,4 +387,25 @@ public interface EnsemblePlacementPolicy {
return (currentStickyBookieIndex.get() + 1) %
metadata.getEnsembleSize();
}
}
+
+ /**
+ * returns true if the Ensemble is strictly adhering to placement policy,
+ * like in the case of RackawareEnsemblePlacementPolicy, bookies in the
+ * writeset are from 'minNumRacksPerWriteQuorum' number of racks. And in
the
+ * case of RegionawareEnsemblePlacementPolicy, check for
+ * minimumRegionsForDurability, reppRegionsToWrite, rack distribution
within
+ * a region and other parameters of RegionAwareEnsemblePlacementPolicy.
+ *
+ * @param ensembleList
+ * list of BookieSocketAddress of bookies in the ensemble
+ * @param writeQuorumSize
+ * writeQuorumSize of the ensemble
+ * @param ackQuorumSize
+ * ackQuorumSize of the ensemble
+ * @return
+ */
+ default boolean
isEnsembleAdheringToPlacementPolicy(List<BookieSocketAddress> ensembleList, int
writeQuorumSize,
+ int ackQuorumSize) {
+ return false;
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ITopologyAwareEnsemblePlacementPolicy.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ITopologyAwareEnsemblePlacementPolicy.java
index 7c9e07c..254f535 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ITopologyAwareEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ITopologyAwareEnsemblePlacementPolicy.java
@@ -28,6 +28,7 @@ import
org.apache.bookkeeper.common.annotation.InterfaceAudience;
import org.apache.bookkeeper.common.annotation.InterfaceStability;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.Node;
+import org.apache.commons.lang3.tuple.Pair;
/**
* Interface for topology aware ensemble placement policy.
@@ -93,7 +94,7 @@ public interface ITopologyAwareEnsemblePlacementPolicy<T
extends Node> extends E
* @return list of bookies forming the ensemble
* @throws BKException.BKNotEnoughBookiesException
*/
- List<BookieSocketAddress> newEnsemble(
+ Pair<List<BookieSocketAddress>, Boolean> newEnsemble(
int ensembleSize,
int writeQuorumSize,
int ackQuorumSize,
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicy.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicy.java
index 1fd7580..8054d97 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicy.java
@@ -18,19 +18,16 @@
package org.apache.bookkeeper.client;
import io.netty.util.HashedWheelTimer;
-
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException;
-import
org.apache.bookkeeper.client.ITopologyAwareEnsemblePlacementPolicy.Ensemble;
-import
org.apache.bookkeeper.client.ITopologyAwareEnsemblePlacementPolicy.Predicate;
-import
org.apache.bookkeeper.client.TopologyAwareEnsemblePlacementPolicy.BookieNode;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.net.Node;
import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.commons.lang3.tuple.Pair;
/**
* A placement policy implementation use rack information for placing
ensembles.
@@ -95,8 +92,8 @@ public class RackawareEnsemblePlacementPolicy extends
RackawareEnsemblePlacement
}
@Override
- public List<BookieSocketAddress> newEnsemble(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
excludeBookies)
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize, int writeQuorumSize,
+ int ackQuorumSize, Map<String, byte[]> customMetadata,
Set<BookieSocketAddress> excludeBookies)
throws BKException.BKNotEnoughBookiesException {
try {
return super.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, customMetadata, excludeBookies);
@@ -110,8 +107,8 @@ public class RackawareEnsemblePlacementPolicy extends
RackawareEnsemblePlacement
}
@Override
- public BookieSocketAddress replaceBookie(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
currentEnsemble,
+ public Pair<BookieSocketAddress, Boolean> replaceBookie(int ensembleSize,
int writeQuorumSize, int ackQuorumSize,
+ Map<String, byte[]> customMetadata, List<BookieSocketAddress>
currentEnsemble,
BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
throws BKException.BKNotEnoughBookiesException {
try {
@@ -146,7 +143,7 @@ public class RackawareEnsemblePlacementPolicy extends
RackawareEnsemblePlacement
}
@Override
- public List<BookieSocketAddress> newEnsemble(int ensembleSize,
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize,
int writeQuorumSize,
int ackQuorumSize,
Set<BookieSocketAddress>
excludeBookies,
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
index 7578c41..6db7de8 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.client;
import static com.google.common.base.Preconditions.checkNotNull;
import static
org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIES_JOINED;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.BOOKIES_LEFT;
+import static
org.apache.bookkeeper.bookie.BookKeeperServerStats.FAILED_TO_RESOLVE_NETWORK_LOCATION_COUNTER;
import static org.apache.bookkeeper.client.BookKeeperClientStats.CLIENT_SCOPE;
import static
org.apache.bookkeeper.client.BookKeeperClientStats.READ_REQUESTS_REORDERED;
import static
org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy.UNKNOWN_REGION;
@@ -35,6 +36,7 @@ import io.netty.util.HashedWheelTimer;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -64,10 +66,12 @@ import org.apache.bookkeeper.net.Node;
import org.apache.bookkeeper.net.NodeBase;
import org.apache.bookkeeper.net.ScriptBasedMapping;
import org.apache.bookkeeper.net.StabilizeNetworkTopology;
+import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.annotations.StatsDoc;
import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -146,12 +150,19 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
final Supplier<String> defaultRackSupplier;
final DNSToSwitchMapping resolver;
-
- DNSResolverDecorator(DNSToSwitchMapping resolver, Supplier<String>
defaultRackSupplier) {
+ @StatsDoc(
+ name = FAILED_TO_RESOLVE_NETWORK_LOCATION_COUNTER,
+ help = "total number of times Resolver failed to resolve rack
information of a node"
+ )
+ final Counter failedToResolveNetworkLocationCounter;
+
+ DNSResolverDecorator(DNSToSwitchMapping resolver, Supplier<String>
defaultRackSupplier,
+ Counter failedToResolveNetworkLocationCounter) {
checkNotNull(resolver, "Resolver cannot be null");
checkNotNull(defaultRackSupplier, "defaultRackSupplier should not
be null");
this.defaultRackSupplier = defaultRackSupplier;
this.resolver = resolver;
+ this.failedToResolveNetworkLocationCounter =
failedToResolveNetworkLocationCounter;
}
public List<String> resolve(List<String> names) {
@@ -167,6 +178,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
if (rNames.get(i) == null) {
LOG.warn("Failed to resolve network location for {},
using default rack for it : {}.",
names.get(i), defaultRack);
+ failedToResolveNetworkLocationCounter.inc();
rNames.set(i, defaultRack);
}
}
@@ -178,6 +190,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
rNames = new ArrayList<>(names.size());
for (int i = 0; i < names.size(); ++i) {
+ failedToResolveNetworkLocationCounter.inc();
rNames.add(defaultRack);
}
return rNames;
@@ -227,6 +240,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
help = "The distribution of number of bookies reordered on each read
request"
)
protected OpStatsLogger readReorderedCounter = null;
+ protected Counter failedToResolveNetworkLocationCounter = null;
private String defaultRack = NetworkTopology.DEFAULT_RACK;
@@ -267,10 +281,13 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
this.bookiesJoinedCounter =
statsLogger.getOpStatsLogger(BOOKIES_JOINED);
this.bookiesLeftCounter = statsLogger.getOpStatsLogger(BOOKIES_LEFT);
this.readReorderedCounter =
statsLogger.getOpStatsLogger(READ_REQUESTS_REORDERED);
+ this.failedToResolveNetworkLocationCounter = statsLogger
+ .getCounter(FAILED_TO_RESOLVE_NETWORK_LOCATION_COUNTER);
this.reorderReadsRandom = reorderReadsRandom;
this.stabilizePeriodSeconds = stabilizePeriodSeconds;
this.reorderThresholdPendingRequests = reorderThresholdPendingRequests;
- this.dnsResolver = new DNSResolverDecorator(dnsResolver, () ->
this.getDefaultRack());
+ this.dnsResolver = new DNSResolverDecorator(dnsResolver, () ->
this.getDefaultRack(),
+ failedToResolveNetworkLocationCounter);
this.timer = timer;
this.minNumRacksPerWriteQuorum = minNumRacksPerWriteQuorum;
this.enforceMinNumRacksPerWriteQuorum =
enforceMinNumRacksPerWriteQuorum;
@@ -342,9 +359,18 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
((RackChangeNotifier)
dnsResolver).registerRackChangeListener(this);
}
} catch (RuntimeException re) {
- LOG.info("Failed to initialize DNS Resolver {}, used default
subnet resolver : {}",
- dnsResolverName, re, re.getMessage());
- dnsResolver = new DefaultResolver(() -> this.getDefaultRack());
+ if (!enforceMinNumRacksPerWriteQuorum) {
+ LOG.info("Failed to initialize DNS Resolver {}, used
default subnet resolver : {}", dnsResolverName,
+ re, re.getMessage());
+ dnsResolver = new DefaultResolver(() ->
this.getDefaultRack());
+ } else {
+ /*
+ * if minNumRacksPerWriteQuorum is enforced, then it
+ * shouldn't continue in the case of failure to create
+ * dnsResolver.
+ */
+ throw re;
+ }
}
}
slowBookies = CacheBuilder.newBuilder()
@@ -479,7 +505,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
}
}
- protected Set<Node> convertBookiesToNodes(Set<BookieSocketAddress>
excludeBookies) {
+ protected Set<Node> convertBookiesToNodes(Collection<BookieSocketAddress>
excludeBookies) {
Set<Node> nodes = new HashSet<Node>();
for (BookieSocketAddress addr : excludeBookies) {
BookieNode bn = knownBookies.get(addr);
@@ -500,13 +526,13 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
}
@Override
- public List<BookieSocketAddress> newEnsemble(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
excludeBookies)
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize, int writeQuorumSize,
+ int ackQuorumSize, Map<String, byte[]> customMetadata,
Set<BookieSocketAddress> excludeBookies)
throws BKNotEnoughBookiesException {
return newEnsembleInternal(ensembleSize, writeQuorumSize,
excludeBookies, null, null);
}
- protected List<BookieSocketAddress> newEnsembleInternal(int ensembleSize,
+ protected Pair<List<BookieSocketAddress>, Boolean> newEnsembleInternal(int
ensembleSize,
int
writeQuorumSize,
Set<BookieSocketAddress> excludeBookies,
Ensemble<BookieNode> parentEnsemble,
@@ -522,7 +548,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
}
@Override
- public List<BookieSocketAddress> newEnsemble(int ensembleSize,
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize,
int writeQuorumSize,
int ackQuorumSize,
Set<BookieSocketAddress>
excludeBookies,
@@ -538,7 +564,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
parentPredicate);
}
- protected List<BookieSocketAddress> newEnsembleInternal(
+ protected Pair<List<BookieSocketAddress>, Boolean> newEnsembleInternal(
int ensembleSize,
int writeQuorumSize,
int ackQuorumSize,
@@ -572,7 +598,7 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
for (BookieNode bn : bns) {
addrs.add(bn.getAddr());
}
- return addrs;
+ return Pair.of(addrs, false);
}
for (int i = 0; i < ensembleSize; i++) {
@@ -596,15 +622,15 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
ensembleSize, bookieList);
throw new BKNotEnoughBookiesException();
}
- return bookieList;
+ return Pair.of(bookieList,
isEnsembleAdheringToPlacementPolicy(bookieList, writeQuorumSize,
ackQuorumSize));
} finally {
rwLock.readLock().unlock();
}
}
@Override
- public BookieSocketAddress replaceBookie(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
currentEnsemble,
+ public Pair<BookieSocketAddress, Boolean> replaceBookie(int ensembleSize,
int writeQuorumSize, int ackQuorumSize,
+ Map<String, byte[]> customMetadata, List<BookieSocketAddress>
currentEnsemble,
BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
throws BKNotEnoughBookiesException {
rwLock.readLock().lock();
@@ -639,7 +665,19 @@ public class RackawareEnsemblePlacementPolicyImpl extends
TopologyAwareEnsembleP
if (LOG.isDebugEnabled()) {
LOG.debug("Bookie {} is chosen to replace bookie {}.",
candidate, bn);
}
- return candidate.getAddr();
+ BookieSocketAddress candidateAddr = candidate.getAddr();
+ List<BookieSocketAddress> newEnsemble = new
ArrayList<BookieSocketAddress>(currentEnsemble);
+ if (currentEnsemble.isEmpty()) {
+ /*
+ * in testing code there are test cases which would pass empty
+ * currentEnsemble
+ */
+ newEnsemble.add(candidateAddr);
+ } else {
+ newEnsemble.set(currentEnsemble.indexOf(bookieToReplace),
candidateAddr);
+ }
+ return Pair.of(candidateAddr,
+ isEnsembleAdheringToPlacementPolicy(newEnsemble,
writeQuorumSize, ackQuorumSize));
} finally {
rwLock.readLock().unlock();
}
@@ -1206,4 +1244,32 @@ public class RackawareEnsemblePlacementPolicyImpl
extends TopologyAwareEnsembleP
}
}
}
+
+ @Override
+ public boolean
isEnsembleAdheringToPlacementPolicy(List<BookieSocketAddress> ensembleList, int
writeQuorumSize,
+ int ackQuorumSize) {
+ int ensembleSize = ensembleList.size();
+ int minNumRacksPerWriteQuorumForThisEnsemble =
Math.min(writeQuorumSize, minNumRacksPerWriteQuorum);
+ HashSet<String> racksOrRegionsInQuorum = new HashSet<String>();
+ BookieSocketAddress bookie;
+ for (int i = 0; i < ensembleList.size(); i++) {
+ racksOrRegionsInQuorum.clear();
+ for (int j = 0; j < writeQuorumSize; j++) {
+ bookie = ensembleList.get((i + j) % ensembleSize);
+ try {
+
racksOrRegionsInQuorum.add(knownBookies.get(bookie).getNetworkLocation());
+ } catch (Exception e) {
+ /*
+ * any issue/exception in analyzing whether ensemble is
strictly adhering to
+ * placement policy should be swallowed.
+ */
+ LOG.warn("Received exception while trying to get network
location of bookie: {}", bookie, e);
+ }
+ }
+ if (racksOrRegionsInQuorum.size() <
minNumRacksPerWriteQuorumForThisEnsemble) {
+ return false;
+ }
+ }
+ return true;
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RegionAwareEnsemblePlacementPolicy.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RegionAwareEnsemblePlacementPolicy.java
index c52e0fe..1bd4b75 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RegionAwareEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RegionAwareEnsemblePlacementPolicy.java
@@ -225,8 +225,8 @@ public class RegionAwareEnsemblePlacementPolicy extends
RackawareEnsemblePlaceme
@Override
- public List<BookieSocketAddress> newEnsemble(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
excludeBookies)
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize, int writeQuorumSize,
+ int ackQuorumSize, Map<String, byte[]> customMetadata,
Set<BookieSocketAddress> excludeBookies)
throws BKException.BKNotEnoughBookiesException {
int effectiveMinRegionsForDurability =
disableDurabilityFeature.isAvailable() ? 1 : minRegionsForDurability;
@@ -279,7 +279,7 @@ public class RegionAwareEnsemblePlacementPolicy extends
RackawareEnsemblePlaceme
for (BookieNode bn : bns) {
addrs.add(bn.getAddr());
}
- return addrs;
+ return Pair.of(addrs,
isEnsembleAdheringToPlacementPolicy(addrs, writeQuorumSize, ackQuorumSize));
}
// Single region, fall back to RackAwareEnsemblePlacement
@@ -347,7 +347,7 @@ public class RegionAwareEnsemblePlacementPolicy extends
RackawareEnsemblePlaceme
try {
List<BookieSocketAddress> allocated =
policyWithinRegion.newEnsemble(newEnsembleSize,
newWriteQuorumSize,
newWriteQuorumSize, excludeBookies, tempEnsemble,
- tempEnsemble);
+ tempEnsemble).getLeft();
ensemble = tempEnsemble;
remainingEnsemble -= addToEnsembleSize;
remainingWriteQuorum -= addToWriteQuorum;
@@ -407,15 +407,17 @@ public class RegionAwareEnsemblePlacementPolicy extends
RackawareEnsemblePlaceme
throw new BKException.BKNotEnoughBookiesException();
}
LOG.info("Bookies allocated successfully {}", ensemble);
- return ensemble.toList();
+ List<BookieSocketAddress> ensembleList = ensemble.toList();
+ return Pair.of(ensembleList,
+ isEnsembleAdheringToPlacementPolicy(ensembleList,
writeQuorumSize, ackQuorumSize));
} finally {
rwLock.readLock().unlock();
}
}
@Override
- public BookieSocketAddress replaceBookie(int ensembleSize, int
writeQuorumSize, int ackQuorumSize,
- Map<String, byte[]> customMetadata, Set<BookieSocketAddress>
currentEnsemble,
+ public Pair<BookieSocketAddress, Boolean> replaceBookie(int ensembleSize,
int writeQuorumSize, int ackQuorumSize,
+ Map<String, byte[]> customMetadata, List<BookieSocketAddress>
currentEnsemble,
BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
throws BKException.BKNotEnoughBookiesException {
rwLock.readLock().lock();
@@ -469,7 +471,19 @@ public class RegionAwareEnsemblePlacementPolicy extends
RackawareEnsemblePlaceme
if (LOG.isDebugEnabled()) {
LOG.debug("Bookie {} is chosen to replace bookie {}.",
candidate, bookieNodeToReplace);
}
- return candidate.getAddr();
+ BookieSocketAddress candidateAddr = candidate.getAddr();
+ List<BookieSocketAddress> newEnsemble = new
ArrayList<BookieSocketAddress>(currentEnsemble);
+ if (currentEnsemble.isEmpty()) {
+ /*
+ * in testing code there are test cases which would pass empty
+ * currentEnsemble
+ */
+ newEnsemble.add(candidateAddr);
+ } else {
+ newEnsemble.set(currentEnsemble.indexOf(bookieToReplace),
candidateAddr);
+ }
+ return Pair.of(candidateAddr,
+ isEnsembleAdheringToPlacementPolicy(newEnsemble,
writeQuorumSize, ackQuorumSize));
} finally {
rwLock.readLock().unlock();
}
@@ -550,4 +564,16 @@ public class RegionAwareEnsemblePlacementPolicy extends
RackawareEnsemblePlaceme
finalList.addMissingIndices(ensemble.size());
return finalList;
}
+
+ @Override
+ public boolean
isEnsembleAdheringToPlacementPolicy(List<BookieSocketAddress> ensembleList, int
writeQuorumSize,
+ int ackQuorumSize) {
+ /**
+ * TODO: have to implement actual logic for this method for
+ * RegionAwareEnsemblePlacementPolicy. For now return true value.
+ *
+ * - https://github.com/apache/bookkeeper/issues/1898
+ */
+ return true;
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/net/ScriptBasedMapping.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/net/ScriptBasedMapping.java
index 8c8350c..230f66d 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/net/ScriptBasedMapping.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/net/ScriptBasedMapping.java
@@ -25,8 +25,8 @@ import java.util.StringTokenizer;
import org.apache.bookkeeper.util.Shell.ShellCommandExecutor;
import org.apache.commons.configuration.Configuration;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This class implements the {@link DNSToSwitchMapping} interface using a
@@ -129,7 +129,7 @@ public final class ScriptBasedMapping extends
CachedDNSToSwitchMapping {
private static final class RawScriptBasedMapping extends
AbstractDNSToSwitchMapping {
private String scriptName;
private int maxArgs; //max hostnames per call of the script
- private static final Log LOG =
LogFactory.getLog(ScriptBasedMapping.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(ScriptBasedMapping.class);
/**
* Set the configuration and extract the configuration parameters of
interest.
@@ -233,7 +233,7 @@ public final class ScriptBasedMapping extends
CachedDNSToSwitchMapping {
s.execute();
allOutput.append(s.getOutput()).append(" ");
} catch (Exception e) {
- LOG.warn("Exception running " + s, e);
+ LOG.warn("Exception running: {} Exception message: {}", s,
e.getMessage());
return null;
}
loopCount++;
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/GenericEnsemblePlacementPolicyTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/GenericEnsemblePlacementPolicyTest.java
index bb55d0c..205c5f4 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/GenericEnsemblePlacementPolicyTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/GenericEnsemblePlacementPolicyTest.java
@@ -33,6 +33,7 @@ import java.util.Set;
import org.apache.bookkeeper.net.BookieSocketAddress;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -69,8 +70,8 @@ public class GenericEnsemblePlacementPolicyTest extends
BookKeeperClusterTestCas
public static final class CustomEnsemblePlacementPolicy extends
DefaultEnsemblePlacementPolicy {
@Override
- public BookieSocketAddress replaceBookie(int ensembleSize, int
writeQuorumSize,
- int ackQuorumSize, Map<String, byte[]> customMetadata,
Set<BookieSocketAddress> currentEnsemble,
+ public Pair<BookieSocketAddress, Boolean> replaceBookie(int
ensembleSize, int writeQuorumSize,
+ int ackQuorumSize, Map<String, byte[]> customMetadata,
List<BookieSocketAddress> currentEnsemble,
BookieSocketAddress bookieToReplace, Set<BookieSocketAddress>
excludeBookies)
throws BKException.BKNotEnoughBookiesException {
new Exception("replaceBookie " + ensembleSize + "," +
customMetadata).printStackTrace();
@@ -81,7 +82,7 @@ public class GenericEnsemblePlacementPolicyTest extends
BookKeeperClusterTestCas
}
@Override
- public ArrayList<BookieSocketAddress> newEnsemble(int ensembleSize,
int quorumSize,
+ public Pair<List<BookieSocketAddress>, Boolean> newEnsemble(int
ensembleSize, int quorumSize,
int ackQuorumSize, Map<String, byte[]> customMetadata,
Set<BookieSocketAddress> excludeBookies)
throws BKException.BKNotEnoughBookiesException {
assertNotNull(customMetadata);
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicy.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicy.java
index e75e93f..e6cd07b 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicy.java
@@ -52,6 +52,7 @@ import org.apache.bookkeeper.net.NetworkTopology;
import org.apache.bookkeeper.net.Node;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.StaticDNSResolver;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,6 +70,7 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
ClientConfiguration conf = new ClientConfiguration();
BookieSocketAddress addr1, addr2, addr3, addr4;
io.netty.util.HashedWheelTimer timer;
+ final int minNumRacksPerWriteQuorumConfValue = 2;
@Override
protected void setUp() throws Exception {
@@ -80,6 +82,7 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
StaticDNSResolver.addNodeToRack("localhost",
NetworkTopology.DEFAULT_REGION_AND_RACK);
LOG.info("Set up static DNS Resolver.");
conf.setProperty(REPP_DNS_RESOLVER_CLASS,
StaticDNSResolver.class.getName());
+ conf.setMinNumRacksPerWriteQuorum(minNumRacksPerWriteQuorumConfValue);
addr1 = new BookieSocketAddress("127.0.0.2", 3181);
addr2 = new BookieSocketAddress("127.0.0.3", 3181);
addr3 = new BookieSocketAddress("127.0.0.4", 3181);
@@ -564,8 +567,12 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
// replace node under r2
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<>(), addr2, new HashSet<>());
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null, new ArrayList<>(),
+ addr2, new HashSet<>());
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
assertEquals(addr3, replacedBookie);
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
@Test
@@ -589,10 +596,13 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
// replace node under r2
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
excludedAddrs.add(addr1);
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<>(), addr2, excludedAddrs);
-
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null, new ArrayList<>(),
+ addr2, excludedAddrs);
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
assertFalse(addr1.equals(replacedBookie));
assertTrue(addr3.equals(replacedBookie) ||
addr4.equals(replacedBookie));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
@Test
@@ -619,7 +629,7 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
excludedAddrs.add(addr3);
excludedAddrs.add(addr4);
try {
- repp.replaceBookie(1, 1, 1, null, new
HashSet<BookieSocketAddress>(), addr2, excludedAddrs);
+ repp.replaceBookie(1, 1, 1, null, new
ArrayList<BookieSocketAddress>(), addr2, excludedAddrs);
fail("Should throw BKNotEnoughBookiesException when there is not
enough bookies");
} catch (BKNotEnoughBookiesException bnebe) {
// should throw not enou
@@ -628,7 +638,7 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
@Test
public void testReplaceBookieWithEnoughBookiesInSameRackAsEnsemble()
throws Exception {
- BookieSocketAddress addr1 = new BookieSocketAddress("127.0.0.1", 3181);
+ BookieSocketAddress addr1 = new BookieSocketAddress("127.0.0.5", 3181);
BookieSocketAddress addr2 = new BookieSocketAddress("127.0.0.2", 3181);
BookieSocketAddress addr3 = new BookieSocketAddress("127.0.0.3", 3181);
BookieSocketAddress addr4 = new BookieSocketAddress("127.0.0.4", 3181);
@@ -645,15 +655,18 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
// replace node under r2
- Set<BookieSocketAddress> ensembleBookies = new
HashSet<BookieSocketAddress>();
+ List<BookieSocketAddress> ensembleBookies = new
ArrayList<BookieSocketAddress>();
ensembleBookies.add(addr2);
ensembleBookies.add(addr4);
- BookieSocketAddress replacedBookie = repp.replaceBookie(
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(
1, 1, 1 , null,
ensembleBookies,
addr4,
new HashSet<>());
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
assertEquals(addr1, replacedBookie);
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
@Test
@@ -670,10 +683,18 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(3, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
+ ensembleResponse = repp.newEnsemble(3, 2, 2, null, new
HashSet<>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
ensembleResponse.getRight();
assertEquals(0, getNumCoveredWriteQuorums(ensemble, 2,
conf.getMinNumRacksPerWriteQuorum()));
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null, new HashSet<>());
+ assertFalse(isEnsembleAdheringToPlacementPolicy);
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2;
+ ensembleResponse2 = repp.newEnsemble(4, 2, 2, null, new
HashSet<>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy2 =
ensembleResponse2.getRight();
assertEquals(0, getNumCoveredWriteQuorums(ensemble2, 2,
conf.getMinNumRacksPerWriteQuorum()));
+ assertFalse(isEnsembleAdheringToPlacementPolicy2);
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
}
@@ -703,17 +724,19 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr3);
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
-
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
try {
- ensemble = repp.newEnsemble(3, 2, 2, null, new HashSet<>());
+ ensembleResponse = repp.newEnsemble(3, 2, 2, null, new
HashSet<>());
+ ensemble = ensembleResponse.getLeft();
fail("Should get not enough bookies exception since there is only
one rack.");
} catch (BKNotEnoughBookiesException bnebe) {
}
try {
- ensemble = repp.newEnsemble(3, 2, 2, new HashSet<>(),
EnsembleForReplacementWithNoConstraints.INSTANCE,
- TruePredicate.INSTANCE);
+ ensembleResponse = repp.newEnsemble(3, 2, 2, new HashSet<>(),
+ EnsembleForReplacementWithNoConstraints.INSTANCE,
TruePredicate.INSTANCE);
+ ensemble = ensembleResponse.getLeft();
fail("Should get not enough bookies exception since there is only
one rack.");
} catch (BKNotEnoughBookiesException bnebe) {
}
@@ -766,19 +789,27 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
* and there are enough bookies in 3 racks, this newEnsemble calls
should
* succeed.
*/
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
+ boolean isEnsembleAdheringToPlacementPolicy;
int ensembleSize = numOfRacks * numOfBookiesPerRack;
int writeQuorumSize = numOfRacks;
int ackQuorumSize = numOfRacks;
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, null, new HashSet<>());
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, null, new HashSet<>());
+ ensemble = ensembleResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy = ensembleResponse.getRight();
assertEquals("Number of writeQuorum sets covered", ensembleSize,
getNumCoveredWriteQuorums(ensemble, writeQuorumSize,
clientConf.getMinNumRacksPerWriteQuorum()));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, new HashSet<>(),
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, new HashSet<>(),
EnsembleForReplacementWithNoConstraints.INSTANCE,
TruePredicate.INSTANCE);
+ ensemble = ensembleResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy = ensembleResponse.getRight();
assertEquals("Number of writeQuorum sets covered", ensembleSize,
getNumCoveredWriteQuorums(ensemble, writeQuorumSize,
clientConf.getMinNumRacksPerWriteQuorum()));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
@Test
@@ -822,16 +853,24 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
* ensembleSizes (as long as there are enough number of bookies in each
* rack).
*/
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
+ boolean isEnsembleAdheringToPlacementPolicy;
for (int ensembleSize = effectiveMinNumRacksPerWriteQuorum;
ensembleSize < 40; ensembleSize++) {
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, null, new HashSet<>());
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, null, new HashSet<>());
+ ensemble = ensembleResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy = ensembleResponse.getRight();
assertEquals("Number of writeQuorum sets covered", ensembleSize,
getNumCoveredWriteQuorums(ensemble, writeQuorumSize,
clientConf.getMinNumRacksPerWriteQuorum()));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, new HashSet<>(),
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, new HashSet<>(),
EnsembleForReplacementWithNoConstraints.INSTANCE,
TruePredicate.INSTANCE);
+ ensemble = ensembleResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy = ensembleResponse.getRight();
assertEquals("Number of writeQuorum sets covered", ensembleSize,
getNumCoveredWriteQuorums(ensemble, writeQuorumSize,
clientConf.getMinNumRacksPerWriteQuorum()));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
}
@@ -873,12 +912,14 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
* and there are enough bookies in 3 racks, this newEnsemble call
should
* succeed.
*/
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
int ensembleSize = numOfRacks * numOfBookiesPerRack;
int writeQuorumSize = numOfRacks;
int ackQuorumSize = numOfRacks;
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, null, new HashSet<>());
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
ackQuorumSize, null, new HashSet<>());
+ ensemble = ensembleResponse.getLeft();
BookieSocketAddress bookieInEnsembleToBeReplaced = ensemble.get(7);
// get rack of some other bookie
@@ -895,7 +936,7 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
repp.onClusterChanged(bookieSocketAddresses, new
HashSet<BookieSocketAddress>());
try {
repp.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize,
null,
- new HashSet<BookieSocketAddress>(ensemble),
bookieInEnsembleToBeReplaced, new HashSet<>());
+ ensemble, bookieInEnsembleToBeReplaced, new HashSet<>());
fail("Should get not enough bookies exception since there are no
more bookies in rack"
+ "of 'bookieInEnsembleToReplace'"
+ "and new bookie added belongs to the rack of some other
bookie in the ensemble");
@@ -917,16 +958,22 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
* this replaceBookie should succeed, because a new bookie is added to
a
* new rack.
*/
- BookieSocketAddress replacedBookieAddress =
repp.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize,
- null, new HashSet<BookieSocketAddress>(ensemble),
bookieInEnsembleToBeReplaced, new HashSet<>());
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse;
+ BookieSocketAddress replacedBookieAddress;
+ boolean isEnsembleAdheringToPlacementPolicy;
+ replaceBookieResponse = repp.replaceBookie(ensembleSize,
writeQuorumSize, ackQuorumSize, null, ensemble,
+ bookieInEnsembleToBeReplaced, new HashSet<>());
+ replacedBookieAddress = replaceBookieResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getRight();
assertEquals("It should be newBookieAddress2", newBookieAddress2,
replacedBookieAddress);
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
Set<BookieSocketAddress> bookiesToExclude = new HashSet<>();
bookiesToExclude.add(newBookieAddress2);
repp.onClusterChanged(bookieSocketAddresses, new
HashSet<BookieSocketAddress>());
try {
- repp.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize,
null,
- new HashSet<BookieSocketAddress>(ensemble),
bookieInEnsembleToBeReplaced, bookiesToExclude);
+ repp.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize,
null, ensemble,
+ bookieInEnsembleToBeReplaced, bookiesToExclude);
fail("Should get not enough bookies exception since the only
available bookie to replace"
+ "is added to excludedBookies list");
} catch (BKNotEnoughBookiesException bnebe) {
@@ -949,9 +996,12 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
* replaced, so we should be able to replacebookie though
* newBookieAddress2 is added to excluded bookies list.
*/
- replacedBookieAddress = repp.replaceBookie(ensembleSize,
writeQuorumSize, ackQuorumSize, null,
- new HashSet<BookieSocketAddress>(ensemble),
bookieInEnsembleToBeReplaced, bookiesToExclude);
+ replaceBookieResponse = repp.replaceBookie(ensembleSize,
writeQuorumSize, ackQuorumSize, null,
+ ensemble, bookieInEnsembleToBeReplaced, bookiesToExclude);
+ replacedBookieAddress = replaceBookieResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getRight();
assertEquals("It should be newBookieAddress3", newBookieAddress3,
replacedBookieAddress);
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
@Test
@@ -1298,15 +1348,21 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
int ensembleSize = 3;
int writeQuorumSize = 2;
int acqQuorumSize = 2;
- List<BookieSocketAddress> ensemble =
repp.newEnsemble(ensembleSize, writeQuorumSize, acqQuorumSize,
- null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(ensembleSize, writeQuorumSize,
+ acqQuorumSize, null, new HashSet<>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
ensembleResponse.getRight();
int numCovered = getNumCoveredWriteQuorums(ensemble,
writeQuorumSize, conf.getMinNumRacksPerWriteQuorum());
assertTrue(numCovered >= 1 && numCovered < 3);
+ assertFalse(isEnsembleAdheringToPlacementPolicy);
ensembleSize = 4;
- List<BookieSocketAddress> ensemble2 =
repp.newEnsemble(ensembleSize, writeQuorumSize, acqQuorumSize,
- null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(ensembleSize, writeQuorumSize,
+ acqQuorumSize, null, new HashSet<>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy2 =
ensembleResponse2.getRight();
numCovered = getNumCoveredWriteQuorums(ensemble2, writeQuorumSize,
conf.getMinNumRacksPerWriteQuorum());
assertTrue(numCovered >= 1 && numCovered < 3);
+ assertFalse(isEnsembleAdheringToPlacementPolicy2);
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
}
@@ -1375,11 +1431,13 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
repp.initialize(newConf, Optional.<DNSToSwitchMapping> empty(), timer,
DISABLE_ALL, NullStatsLogger.INSTANCE);
repp.withDefaultRack(NetworkTopology.DEFAULT_REGION_AND_RACK);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
-
- List<BookieSocketAddress> ensemble = repp.newEnsemble(ensembleSize,
writeQuorumSize, writeQuorumSize, null,
- new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(ensembleSize, writeQuorumSize,
+ writeQuorumSize, null, new HashSet<>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
ensembleResponse.getRight();
int numCovered = getNumCoveredWriteQuorums(ensemble, writeQuorumSize,
minNumRacksPerWriteQuorum);
assertEquals("minimum number of racks covered for writequorum
ensemble: " + ensemble, ensembleSize, numCovered);
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
}
@Test
@@ -1417,16 +1475,22 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
int ensembleSize = 3;
int writeQuorumSize = 3;
int ackQuorumSize = 2;
- List<BookieSocketAddress> ensemble1 =
repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize,
- null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(ensembleSize, writeQuorumSize,
+ ackQuorumSize, null, new HashSet<>());
+ List<BookieSocketAddress> ensemble1 = ensembleResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy1 =
ensembleResponse.getRight();
assertEquals(ensembleSize,
getNumCoveredWriteQuorums(ensemble1, writeQuorumSize,
conf.getMinNumRacksPerWriteQuorum()));
+ assertTrue(isEnsembleAdheringToPlacementPolicy1);
ensembleSize = 4;
writeQuorumSize = 4;
- List<BookieSocketAddress> ensemble2 =
repp.newEnsemble(ensembleSize, writeQuorumSize, 2, null,
- new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(ensembleSize, writeQuorumSize,
+ 2, null, new HashSet<>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy2 =
ensembleResponse2.getRight();
assertEquals(ensembleSize,
getNumCoveredWriteQuorums(ensemble2, writeQuorumSize,
conf.getMinNumRacksPerWriteQuorum()));
+ assertTrue(isEnsembleAdheringToPlacementPolicy2);
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
}
@@ -1497,11 +1561,16 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
selectionCounts.put(addr3, 0L);
selectionCounts.put(addr4, 0L);
int numTries = 50000;
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse;
+ boolean isEnsembleAdheringToPlacementPolicy;
BookieSocketAddress replacedBookie;
for (int i = 0; i < numTries; i++) {
// replace node under r2
- replacedBookie = repp.replaceBookie(1, 1, 1, null, new
HashSet<>(), addr2, new HashSet<>());
+ replaceBookieResponse = repp.replaceBookie(1, 1, 1, null, new
ArrayList<>(), addr2, new HashSet<>());
+ replacedBookie = replaceBookieResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
assertTrue("replaced : " + replacedBookie,
addr3.equals(replacedBookie) || addr4.equals(replacedBookie));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
selectionCounts.put(replacedBookie,
selectionCounts.get(replacedBookie) + 1);
}
double observedMultiple = ((double) selectionCounts.get(addr4) /
(double) selectionCounts.get(addr3));
@@ -1557,14 +1626,19 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
selectionCounts.put(addr3, 0L);
selectionCounts.put(addr4, 0L);
int numTries = 50000;
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse;
BookieSocketAddress replacedBookie;
+ boolean isEnsembleAdheringToPlacementPolicy;
for (int i = 0; i < numTries; i++) {
// addr2 is on /r2 and this is the only one on this rack. So the
replacement
// will come from other racks. However, the weight should be
honored in such
// selections as well
- replacedBookie = repp.replaceBookie(1, 1, 1, null, new
HashSet<>(), addr2, new HashSet<>());
+ replaceBookieResponse = repp.replaceBookie(1, 1, 1, null, new
ArrayList<>(), addr2, new HashSet<>());
+ replacedBookie = replaceBookieResponse.getLeft();
+ isEnsembleAdheringToPlacementPolicy =
replaceBookieResponse.getRight();
assertTrue(addr0.equals(replacedBookie) ||
addr1.equals(replacedBookie) || addr3.equals(replacedBookie)
|| addr4.equals(replacedBookie));
+ assertTrue(isEnsembleAdheringToPlacementPolicy);
selectionCounts.put(replacedBookie,
selectionCounts.get(replacedBookie) + 1);
}
/*
@@ -1656,6 +1730,7 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
int numTries = 10000;
Set<BookieSocketAddress> excludeList = new
HashSet<BookieSocketAddress>();
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
int ensembleSize = 3;
int writeQuorumSize = 2;
@@ -1664,7 +1739,8 @@ public class TestRackawareEnsemblePlacementPolicy extends
TestCase {
// addr2 is on /r2 and this is the only one on this rack. So the
replacement
// will come from other racks. However, the weight should be
honored in such
// selections as well
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
acqQuorumSize, null, excludeList);
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
acqQuorumSize, null, excludeList);
+ ensemble = ensembleResponse.getLeft();
assertTrue(
"Rackaware selection not happening "
+ getNumCoveredWriteQuorums(ensemble,
writeQuorumSize, conf.getMinNumRacksPerWriteQuorum()),
@@ -1726,21 +1802,23 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
bookieInfoMap.put(addr5, new BookieInfo(1000L, 1000L));
repp.updateBookieInfo(bookieInfoMap);
-
- List<BookieSocketAddress> ensemble = new
ArrayList<BookieSocketAddress>();
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
+ List<BookieSocketAddress> ensemble;
Set<BookieSocketAddress> excludeList = new
HashSet<BookieSocketAddress>();
try {
excludeList.add(addr1);
excludeList.add(addr2);
excludeList.add(addr3);
excludeList.add(addr4);
- ensemble = repp.newEnsemble(3, 2, 2, null, excludeList);
+ ensembleResponse = repp.newEnsemble(3, 2, 2, null, excludeList);
+ ensemble = ensembleResponse.getLeft();
fail("Should throw BKNotEnoughBookiesException when there is not
enough bookies" + ensemble);
} catch (BKNotEnoughBookiesException e) {
// this is expected
}
try {
- ensemble = repp.newEnsemble(1, 1, 1, null, excludeList);
+ ensembleResponse = repp.newEnsemble(1, 1, 1, null, excludeList);
+ ensemble = ensembleResponse.getLeft();
} catch (BKNotEnoughBookiesException e) {
fail("Should not throw BKNotEnoughBookiesException when there are
enough bookies for the ensemble");
}
@@ -1824,13 +1902,20 @@ public class TestRackawareEnsemblePlacementPolicy
extends TestCase {
// we will never use addr4 even it is in the stabilized network
topology
for (int i = 0; i < 5; i++) {
- List<BookieSocketAddress> ensemble =
- repp.newEnsemble(3, 3, 3, null, new
HashSet<BookieSocketAddress>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(3, 2, 2, null,
+ new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
ensembleResponse.getRight();
assertFalse(ensemble.contains(addr4));
+ assertFalse(isEnsembleAdheringToPlacementPolicy);
}
// we could still use addr4 for urgent allocation if it is just bookie
flapping
- List<BookieSocketAddress> ensemble = repp.newEnsemble(4, 4, 4, null,
new HashSet<BookieSocketAddress>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(4, 2, 2, null,
+ new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
+ boolean isEnsembleAdheringToPlacementPolicy =
ensembleResponse.getRight();
+ assertFalse(isEnsembleAdheringToPlacementPolicy);
assertTrue(ensemble.contains(addr4));
}
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicyUsingScript.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicyUsingScript.java
index d9f2535..e0fd2bd 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicyUsingScript.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicyUsingScript.java
@@ -30,6 +30,8 @@ import static org.junit.Assert.fail;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.util.HashedWheelTimer;
+
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
@@ -44,6 +46,7 @@ import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.net.ScriptBasedMapping;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.Shell;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
@@ -111,7 +114,9 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
// replace node under r2
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<>(), addr2, new HashSet<>());
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null, new ArrayList<>(),
+ addr2, new HashSet<>());
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
assertEquals(addr3, replacedBookie);
}
@@ -133,7 +138,9 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
// replace node under r2
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
excludedAddrs.add(addr1);
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<>(), addr2, excludedAddrs);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null, new ArrayList<>(),
+ addr2, excludedAddrs);
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
assertFalse(addr1.equals(replacedBookie));
assertTrue(addr3.equals(replacedBookie) ||
addr4.equals(replacedBookie));
@@ -160,7 +167,7 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
excludedAddrs.add(addr3);
excludedAddrs.add(addr4);
try {
- repp.replaceBookie(1, 1, 1, null, new
HashSet<BookieSocketAddress>(), addr2, excludedAddrs);
+ repp.replaceBookie(1, 1, 1, null, new
ArrayList<BookieSocketAddress>(), addr2, excludedAddrs);
fail("Should throw BKNotEnoughBookiesException when there is not
enough bookies");
} catch (BKNotEnoughBookiesException bnebe) {
// should throw not BKNotEnoughBookiesException
@@ -197,7 +204,9 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
// replace node under r2
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
excludedAddrs.add(addr1);
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<>(), addr2, excludedAddrs);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null, new ArrayList<>(),
+ addr2, excludedAddrs);
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
assertFalse(addr1.equals(replacedBookie));
assertFalse(addr2.equals(replacedBookie));
@@ -235,7 +244,9 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
// replace node under r2
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
excludedAddrs.add(addr1);
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<>(), addr2, excludedAddrs);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null, new ArrayList<>(),
+ addr2, excludedAddrs);
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
assertFalse(addr1.equals(replacedBookie));
assertFalse(addr2.equals(replacedBookie));
@@ -257,9 +268,13 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(3, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(3, 2, 2, null,
+ new HashSet<>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assertEquals(0, getNumCoveredWriteQuorums(ensemble, 2));
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(4, 2, 2, null,
+ new HashSet<>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
assertEquals(0, getNumCoveredWriteQuorums(ensemble2, 2));
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
@@ -281,10 +296,14 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(3, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(3, 2, 2, null,
+ new HashSet<>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
int numCovered = getNumCoveredWriteQuorums(ensemble, 2);
assertTrue(numCovered == 2);
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(4, 2, 2, null,
+ new HashSet<>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
numCovered = getNumCoveredWriteQuorums(ensemble2, 2);
assertTrue(numCovered == 2);
} catch (BKNotEnoughBookiesException bnebe) {
@@ -315,9 +334,13 @@ public class
TestRackawareEnsemblePlacementPolicyUsingScript {
addrs.add(addr8);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble1 = repp.newEnsemble(3, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse1 =
repp.newEnsemble(3, 2, 2, null,
+ new HashSet<>());
+ List<BookieSocketAddress> ensemble1 = ensembleResponse1.getLeft();
assertEquals(3, getNumCoveredWriteQuorums(ensemble1, 2));
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null, new HashSet<>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(4, 2, 2, null,
+ new HashSet<>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
assertEquals(4, getNumCoveredWriteQuorums(ensemble2, 2));
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception.");
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawarePolicyNotificationUpdates.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawarePolicyNotificationUpdates.java
index 7dc1d39..3192d04 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawarePolicyNotificationUpdates.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawarePolicyNotificationUpdates.java
@@ -42,6 +42,7 @@ import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.net.NetworkTopology;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.StaticDNSResolver;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -104,8 +105,9 @@ public class TestRackawarePolicyNotificationUpdates extends
TestCase {
int ensembleSize = 3;
int writeQuorumSize = 2;
int acqQuorumSize = 2;
- List<BookieSocketAddress> ensemble = repp.newEnsemble(ensembleSize,
writeQuorumSize, acqQuorumSize,
- Collections.emptyMap(), Collections.emptySet());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(ensembleSize, writeQuorumSize,
+ acqQuorumSize, Collections.emptyMap(), Collections.emptySet());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
int numCovered =
TestRackawareEnsemblePlacementPolicy.getNumCoveredWriteQuorums(ensemble,
writeQuorumSize,
conf.getMinNumRacksPerWriteQuorum());
assertTrue(numCovered >= 1 && numCovered < 3);
@@ -118,8 +120,9 @@ public class TestRackawarePolicyNotificationUpdates extends
TestCase {
StaticDNSResolver.changeRack(bookieAddressList, rackList);
numOfAvailableRacks = numOfAvailableRacks + 1;
acqQuorumSize = 1;
- ensemble = repp.newEnsemble(ensembleSize, writeQuorumSize,
acqQuorumSize, Collections.emptyMap(),
+ ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize,
acqQuorumSize, Collections.emptyMap(),
Collections.emptySet());
+ ensemble = ensembleResponse.getLeft();
assertEquals(3,
TestRackawareEnsemblePlacementPolicy.getNumCoveredWriteQuorums(ensemble,
writeQuorumSize,
conf.getMinNumRacksPerWriteQuorum()));
assertTrue(ensemble.contains(addr1));
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRegionAwareEnsemblePlacementPolicy.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRegionAwareEnsemblePlacementPolicy.java
index e541230..8e4f10d 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRegionAwareEnsemblePlacementPolicy.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRegionAwareEnsemblePlacementPolicy.java
@@ -51,6 +51,7 @@ import org.apache.bookkeeper.net.NetworkTopology;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.util.StaticDNSResolver;
+import org.apache.commons.lang3.tuple.Pair;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -423,8 +424,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
// replace node under r2
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
new HashSet<BookieSocketAddress>(),
- addr2, new HashSet<BookieSocketAddress>());
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null,
+ new ArrayList<BookieSocketAddress>(), addr2, new
HashSet<BookieSocketAddress>());
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
assertEquals(addr3, replacedBookie);
}
@@ -449,8 +451,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
// replace node under r2
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
excludedAddrs.add(addr1);
- BookieSocketAddress replacedBookie = repp.replaceBookie(1, 1, 1, null,
- new HashSet<BookieSocketAddress>(), addr2, excludedAddrs);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(1, 1, 1, null,
+ new ArrayList<BookieSocketAddress>(), addr2, excludedAddrs);
+ BookieSocketAddress replacedBookie = replaceBookieResponse.getLeft();
assertFalse(addr1.equals(replacedBookie));
assertTrue(addr3.equals(replacedBookie) ||
addr4.equals(replacedBookie));
@@ -475,7 +478,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> list = repp.newEnsemble(5, 5, 3, null,
new HashSet<BookieSocketAddress>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(5, 5, 3, null,
+ new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> list = ensembleResponse.getLeft();
LOG.info("Ensemble : {}", list);
fail("Should throw BKNotEnoughBookiesException when there is not
enough bookies");
} catch (BKNotEnoughBookiesException bnebe) {
@@ -507,7 +512,7 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
excludedAddrs.add(addr3);
excludedAddrs.add(addr4);
try {
- repp.replaceBookie(1, 1, 1, null, new
HashSet<BookieSocketAddress>(), addr2, excludedAddrs);
+ repp.replaceBookie(1, 1, 1, null, new
ArrayList<BookieSocketAddress>(), addr2, excludedAddrs);
fail("Should throw BKNotEnoughBookiesException when there is not
enough bookies");
} catch (BKNotEnoughBookiesException bnebe) {
// should throw not enou
@@ -536,11 +541,13 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(3, 2, 2,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(3, 2, 2, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assertEquals(0, getNumCoveredRegionsInWriteQuorum(ensemble, 2));
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(4, 2, 2, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
assertEquals(0, getNumCoveredRegionsInWriteQuorum(ensemble2, 2));
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
@@ -569,8 +576,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
addrs.add(addr4);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(3, 2, 2,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(3, 2, 2, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
int numCovered = getNumCoveredRegionsInWriteQuorum(ensemble, 2);
assertTrue(numCovered >= 1);
assertTrue(numCovered < 3);
@@ -578,8 +586,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
fail("Should not get not enough bookies exception even there is
only one rack.");
}
try {
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(4, 2, 2, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
int numCovered = getNumCoveredRegionsInWriteQuorum(ensemble2, 2);
assertTrue(numCovered >= 1 && numCovered < 3);
} catch (BKNotEnoughBookiesException bnebe) {
@@ -618,11 +627,13 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
addrs.add(addr8);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble1 = repp.newEnsemble(3, 2, 2,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse1 =
repp.newEnsemble(3, 2, 2, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble1 = ensembleResponse1.getLeft();
assertEquals(3, getNumCoveredRegionsInWriteQuorum(ensemble1, 2));
- List<BookieSocketAddress> ensemble2 = repp.newEnsemble(4, 2, 2,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse2 =
repp.newEnsemble(4, 2, 2, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble2 = ensembleResponse2.getLeft();
assertEquals(4, getNumCoveredRegionsInWriteQuorum(ensemble2, 2));
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
@@ -669,23 +680,27 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
addrs.add(addr10);
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(6, 6, 4,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(6, 6, 4, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assert(ensemble.contains(addr4));
assert(ensemble.contains(addr8));
assert(ensemble.size() == 6);
assertEquals(3, getNumRegionsInEnsemble(ensemble));
- ensemble = repp.newEnsemble(7, 7, 4, null, new
HashSet<BookieSocketAddress>());
+ ensembleResponse = repp.newEnsemble(7, 7, 4, null, new
HashSet<BookieSocketAddress>());
+ ensemble = ensembleResponse.getLeft();
assert(ensemble.contains(addr4));
assert(ensemble.contains(addr8));
assert(ensemble.size() == 7);
assertEquals(3, getNumRegionsInEnsemble(ensemble));
- ensemble = repp.newEnsemble(8, 8, 5, null, new
HashSet<BookieSocketAddress>());
+ ensembleResponse = repp.newEnsemble(8, 8, 5, null, new
HashSet<BookieSocketAddress>());
+ ensemble = ensembleResponse.getLeft();
assert(ensemble.contains(addr4));
assert(ensemble.contains(addr8));
assert(ensemble.size() == 8);
assertEquals(3, getNumRegionsInEnsemble(ensemble));
- ensemble = repp.newEnsemble(9, 9, 5, null, new
HashSet<BookieSocketAddress>());
+ ensembleResponse = repp.newEnsemble(9, 9, 5, null, new
HashSet<BookieSocketAddress>());
+ ensemble = ensembleResponse.getLeft();
assert(ensemble.contains(addr4));
assert(ensemble.contains(addr8));
assert(ensemble.size() == 9);
@@ -738,8 +753,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
((SettableFeature)
featureProvider.scope("region1").getFeature("disallowBookies")).set(true);
- List<BookieSocketAddress> ensemble = repp.newEnsemble(6, 6, 4,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(6, 6, 4, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assertEquals(2, getNumRegionsInEnsemble(ensemble));
assert(ensemble.contains(addr1));
assert(ensemble.contains(addr3));
@@ -753,16 +769,16 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
}
try {
((SettableFeature)
featureProvider.scope("region2").getFeature("disallowBookies")).set(true);
- List<BookieSocketAddress> ensemble = repp.newEnsemble(6, 6, 4,
null,
- new HashSet<BookieSocketAddress>());
+ repp.newEnsemble(6, 6, 4, null, new
HashSet<BookieSocketAddress>());
fail("Should get not enough bookies exception even there is only
one region with insufficient bookies.");
} catch (BKNotEnoughBookiesException bnebe) {
// Expected
}
try {
((SettableFeature)
featureProvider.scope("region2").getFeature("disallowBookies")).set(false);
- List<BookieSocketAddress> ensemble = repp.newEnsemble(6, 6, 4,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(6, 6, 4, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assert(ensemble.contains(addr1));
assert(ensemble.contains(addr3));
assert(ensemble.contains(addr4));
@@ -835,8 +851,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
repp.onClusterChanged(addrs, new HashSet<BookieSocketAddress>());
try {
- List<BookieSocketAddress> ensemble = repp.newEnsemble(10, 10, 10,
null,
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(10, 10, 10, null,
new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assert(ensemble.size() == 10);
assertEquals(5, getNumRegionsInEnsemble(ensemble));
} catch (BKNotEnoughBookiesException bnebe) {
@@ -847,7 +864,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
try {
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
excludedAddrs.add(addr10);
- List<BookieSocketAddress> ensemble = repp.newEnsemble(10, 10, 10,
null, excludedAddrs);
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(10, 10, 10, null,
+ excludedAddrs);
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assert(ensemble.contains(addr11) && ensemble.contains(addr12));
assert(ensemble.size() == 10);
assertEquals(5, getNumRegionsInEnsemble(ensemble));
@@ -937,9 +956,11 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
ackQuorum = 5;
}
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
try {
- ensemble = repp.newEnsemble(6, 6, ackQuorum, null, new
HashSet<BookieSocketAddress>());
+ ensembleResponse = repp.newEnsemble(6, 6, ackQuorum, null, new
HashSet<BookieSocketAddress>());
+ ensemble = ensembleResponse.getLeft();
assert(ensemble.size() == 6);
assertEquals(3, getNumRegionsInEnsemble(ensemble));
} catch (BKNotEnoughBookiesException bnebe) {
@@ -960,9 +981,9 @@ public class TestRegionAwareEnsemblePlacementPolicy extends
TestCase {
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
for (BookieSocketAddress addr: region2Bookies) {
if (ensemble.contains(addr)) {
- BookieSocketAddress replacedBookie = repp.replaceBookie(
- 6, 6, ackQuorum, null,
- new HashSet<>(ensemble), addr, excludedAddrs);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(6, 6, ackQuorum, null,
+ ensemble, addr, excludedAddrs);
+ BookieSocketAddress replacedBookie =
replaceBookieResponse.getLeft();
ensemble.remove(addr);
ensemble.add(replacedBookie);
}
@@ -986,9 +1007,9 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
try {
- BookieSocketAddress replacedBookie = repp.replaceBookie(
- 6, 6, ackQuorum, null,
- new HashSet<>(ensemble), bookieToReplace, excludedAddrs);
+ Pair<BookieSocketAddress, Boolean> replaceBookieResponse =
repp.replaceBookie(6, 6, ackQuorum, null,
+ ensemble, bookieToReplace, excludedAddrs);
+ BookieSocketAddress replacedBookie =
replaceBookieResponse.getLeft();
assert (replacedBookie.equals(replacedBookieExpected));
assertEquals(3, getNumRegionsInEnsemble(ensemble));
} catch (BKNotEnoughBookiesException bnebe) {
@@ -997,9 +1018,7 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
excludedAddrs.add(replacedBookieExpected);
try {
- BookieSocketAddress replacedBookie = repp.replaceBookie(
- 6, 6, ackQuorum, null,
- new HashSet<>(ensemble), bookieToReplace, excludedAddrs);
+ repp.replaceBookie(6, 6, ackQuorum, null, ensemble,
bookieToReplace, excludedAddrs);
if (minDurability > 1 &&
!disableDurabilityFeature.isAvailable()) {
fail("Should throw BKNotEnoughBookiesException when there
is not enough bookies");
}
@@ -1073,9 +1092,11 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
.set(true);
}
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse;
List<BookieSocketAddress> ensemble;
try {
- ensemble = repp.newEnsemble(6, 6, 4, null, new
HashSet<BookieSocketAddress>());
+ ensembleResponse = repp.newEnsemble(6, 6, 4, null, new
HashSet<BookieSocketAddress>());
+ ensemble = ensembleResponse.getLeft();
assert(ensemble.size() == 6);
} catch (BKNotEnoughBookiesException bnebe) {
LOG.error("BKNotEnoughBookiesException", bnebe);
@@ -1086,9 +1107,7 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
Set<BookieSocketAddress> excludedAddrs = new
HashSet<BookieSocketAddress>();
try {
- repp.replaceBookie(
- 6, 6, 4, null,
- new HashSet<>(ensemble), addr4, excludedAddrs);
+ repp.replaceBookie(6, 6, 4, null, ensemble, ensemble.get(2),
excludedAddrs);
} catch (BKNotEnoughBookiesException bnebe) {
fail("Should not get not enough bookies exception even there is
only one rack.");
}
@@ -1141,8 +1160,8 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
excludedAddrs.add(addr10);
excludedAddrs.add(addr9);
try {
- List<BookieSocketAddress> list = repp.newEnsemble(5, 5, 5, null,
excludedAddrs);
- LOG.info("Ensemble : {}", list);
+ Pair<List<BookieSocketAddress>, Boolean> list =
repp.newEnsemble(5, 5, 5, null, excludedAddrs);
+ LOG.info("Ensemble : {}", list.getLeft());
fail("Should throw BKNotEnoughBookiesException when there is not
enough bookies");
} catch (BKNotEnoughBookiesException bnebe) {
// should throw not enou
@@ -1201,8 +1220,9 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
private void basicReorderReadSequenceWithLocalRegionTest(String myRegion,
boolean isReadLAC) throws Exception {
prepareNetworkTopologyForReorderTests(myRegion);
-
- List<BookieSocketAddress> ensemble = repp.newEnsemble(9, 9, 5, null,
new HashSet<BookieSocketAddress>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(9, 9, 5, null,
+ new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assertEquals(9, getNumCoveredRegionsInWriteQuorum(ensemble, 9));
DistributionSchedule ds = new RoundRobinDistributionSchedule(9, 9, 9);
@@ -1258,7 +1278,9 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
private void basicReorderReadSequenceWithRemoteRegionTest(String myRegion,
boolean isReadLAC) throws Exception {
prepareNetworkTopologyForReorderTests(myRegion);
- List<BookieSocketAddress> ensemble = repp.newEnsemble(9, 9, 5, null,
new HashSet<BookieSocketAddress>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(9, 9, 5, null,
+ new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assertEquals(9, getNumCoveredRegionsInWriteQuorum(ensemble, 9));
DistributionSchedule ds = new RoundRobinDistributionSchedule(9, 9, 9);
@@ -1328,7 +1350,9 @@ public class TestRegionAwareEnsemblePlacementPolicy
extends TestCase {
prepareNetworkTopologyForReorderTests(myRegion);
- List<BookieSocketAddress> ensemble = repp.newEnsemble(9, 9, 5, null,
new HashSet<BookieSocketAddress>());
+ Pair<List<BookieSocketAddress>, Boolean> ensembleResponse =
repp.newEnsemble(9, 9, 5, null,
+ new HashSet<BookieSocketAddress>());
+ List<BookieSocketAddress> ensemble = ensembleResponse.getLeft();
assertEquals(9, getNumCoveredRegionsInWriteQuorum(ensemble, 9));
DistributionSchedule ds = new RoundRobinDistributionSchedule(9, 9, 9);
