diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java index f732a97726f..08853e6f955 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java @@ -567,7 +567,7 @@ public BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo bookieQuarantineRatio = 1.0; } - private EnsemblePlacementPolicy initializeEnsemblePlacementPolicy(ClientConfiguration conf, + protected EnsemblePlacementPolicy initializeEnsemblePlacementPolicy(ClientConfiguration conf, DNSToSwitchMapping dnsResolver, HashedWheelTimer timer, FeatureProvider featureProvider, 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 d62b6207914..2e95ab5a4cb 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 @@ -33,8 +33,10 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.Enumeration; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.LinkedList; import java.util.List; @@ -90,6 +92,7 @@ import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.util.AvailabilityOfEntriesOfLedger; import org.apache.bookkeeper.util.IOUtils; +import org.apache.commons.collections4.MapUtils; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.KeeperException; import org.slf4j.Logger; @@ -1095,7 +1098,7 @@ private Map getReplacementBookiesByIndexes( oldBookie, bookiesToExclude); BookieId newBookie = replaceBookieResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); if (isEnsembleAdheringToPlacementPolicy == PlacementPolicyAdherence.FAIL) { if (LOG.isDebugEnabled()) { LOG.debug( @@ -1129,14 +1132,23 @@ private ArrayList replaceBookiesInEnsemble( * @param ledgerFragment * - LedgerFragment to replicate */ - public void replicateLedgerFragment(LedgerHandle lh, - final LedgerFragment ledgerFragment, - final BiConsumer onReadEntryFailureCallback) - throws InterruptedException, BKException { - Optional> excludedBookies = Optional.empty(); - Map targetBookieAddresses = - getReplacementBookiesByIndexes(lh, ledgerFragment.getEnsemble(), - ledgerFragment.getBookiesIndexes(), excludedBookies); + public void replicateLedgerFragment(LedgerHandle lh, final LedgerFragment ledgerFragment, + final BiConsumer onReadEntryFailureCallback) throws InterruptedException, BKException { + Map targetBookieAddresses = null; + if (LedgerFragment.ReplicateType.DATA_LOSS == ledgerFragment.getReplicateType()) { + Optional> excludedBookies = Optional.empty(); + targetBookieAddresses = getReplacementBookiesByIndexes(lh, ledgerFragment.getEnsemble(), + ledgerFragment.getBookiesIndexes(), excludedBookies); + } else if (LedgerFragment.ReplicateType.DATA_NOT_ADHERING_PLACEMENT == ledgerFragment.getReplicateType()) { + targetBookieAddresses = replaceNotAdheringPlacementPolicyBookie(ledgerFragment.getEnsemble(), + lh.getLedgerMetadata().getWriteQuorumSize(), lh.getLedgerMetadata().getAckQuorumSize()); + ledgerFragment.getBookiesIndexes().addAll(targetBookieAddresses.keySet()); + } + if (MapUtils.isEmpty(targetBookieAddresses)) { + LOG.warn("Could not replicate for {} ledger: {}, not find target bookie.", + ledgerFragment.getReplicateType(), ledgerFragment.getLedgerId()); + return; + } replicateLedgerFragment(lh, ledgerFragment, targetBookieAddresses, onReadEntryFailureCallback); } @@ -1777,6 +1789,31 @@ public PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy(List replaceNotAdheringPlacementPolicyBookie(List ensembleBookiesList, + int writeQuorumSize, int ackQuorumSize) { + try { + EnsemblePlacementPolicy.PlacementResult> placementResult = bkc.getPlacementPolicy() + .replaceToAdherePlacementPolicy(ensembleBookiesList.size(), writeQuorumSize, ackQuorumSize, + new HashSet<>(), ensembleBookiesList); + if (PlacementPolicyAdherence.FAIL != placementResult.getAdheringToPolicy()) { + Map targetMap = new HashMap<>(); + List newEnsembles = placementResult.getResult(); + for (int i = 0; i < ensembleBookiesList.size(); i++) { + BookieId originBookie = ensembleBookiesList.get(i); + BookieId newBookie = newEnsembles.get(i); + if (!originBookie.equals(newBookie)) { + targetMap.put(i, newBookie); + } + } + return targetMap; + } + } catch (UnsupportedOperationException e) { + LOG.warn("The placement policy: {} didn't support replaceToAdherePlacementPolicy, " + + "ignore replace not adhere bookie.", bkc.getPlacementPolicy().getClass().getName()); + } + return Collections.emptyMap(); + } + /** * Makes async request for getting list of entries of ledger from a bookie * and returns Future for the result. 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 c2806877aca..652fbcc7733 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 @@ -269,7 +269,7 @@ public List newEnsemble(int ensembleSize, int writeQuorumSize, newEnsembleResponse = placementPolicy.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, customMetadata, new HashSet(quarantinedBookiesSet)); socketAddresses = newEnsembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = newEnsembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = newEnsembleResponse.getAdheringToPolicy(); if (isEnsembleAdheringToPlacementPolicy == PlacementPolicyAdherence.FAIL) { ensembleNotAdheringToPlacementPolicy.inc(); if (ensembleSize > 1) { @@ -286,7 +286,7 @@ public List newEnsemble(int ensembleSize, int writeQuorumSize, newEnsembleResponse = placementPolicy.newEnsemble( ensembleSize, writeQuorumSize, ackQuorumSize, customMetadata, new HashSet<>()); socketAddresses = newEnsembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = newEnsembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = newEnsembleResponse.getAdheringToPolicy(); if (isEnsembleAdheringToPlacementPolicy == PlacementPolicyAdherence.FAIL) { ensembleNotAdheringToPlacementPolicy.inc(); log.warn("New ensemble: {} is not adhering to Placement Policy", socketAddresses); @@ -317,7 +317,7 @@ public BookieId replaceBookie(int ensembleSize, int writeQuorumSize, int ackQuor ensembleSize, writeQuorumSize, ackQuorumSize, customMetadata, existingBookies, addr, excludedBookiesAndQuarantinedBookies); socketAddress = replaceBookieResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); if (isEnsembleAdheringToPlacementPolicy == PlacementPolicyAdherence.FAIL) { ensembleNotAdheringToPlacementPolicy.inc(); log.warn( @@ -333,7 +333,7 @@ public BookieId replaceBookie(int ensembleSize, int writeQuorumSize, int ackQuor replaceBookieResponse = placementPolicy.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize, customMetadata, existingBookies, addr, excludeBookies); socketAddress = replaceBookieResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); if (isEnsembleAdheringToPlacementPolicy == PlacementPolicyAdherence.FAIL) { ensembleNotAdheringToPlacementPolicy.inc(); log.warn( 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 39aec4574ca..05a687b964d 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 @@ -440,6 +440,31 @@ default boolean areAckedBookiesAdheringToPlacementPolicy(Set ackedBook return true; } + /** + * Returns placement result. If the currentEnsemble is not adhering placement policy, returns new ensemble that + * adheres placement policy. It should be implemented so as to minify the number of bookies replaced. + * + * @param ensembleSize + * ensemble size + * @param writeQuorumSize + * writeQuorumSize of the ensemble + * @param ackQuorumSize + * ackQuorumSize of the ensemble + * @param excludeBookies + * bookies that should not be considered as targets + * @param currentEnsemble + * current ensemble + * @return a placement result + */ + default PlacementResult> replaceToAdherePlacementPolicy( + int ensembleSize, + int writeQuorumSize, + int ackQuorumSize, + Set excludeBookies, + List currentEnsemble) { + throw new UnsupportedOperationException(); + } + /** * enum for PlacementPolicyAdherence. Currently we are supporting tri-value * enum for PlacementPolicyAdherence. If placement policy is met strictly @@ -481,8 +506,16 @@ public T getResult() { return result; } + /** + * Use {@link #getAdheringToPolicy}. + */ + @Deprecated public PlacementPolicyAdherence isAdheringToPolicy() { return policyAdherence; } + + public PlacementPolicyAdherence getAdheringToPolicy() { + return policyAdherence; + } } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragment.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragment.java index 1fb1e50cb02..2dffa10f836 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragment.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerFragment.java @@ -39,8 +39,9 @@ public class LedgerFragment { private final long ledgerId; private final DistributionSchedule schedule; private final boolean isLedgerClosed; + private ReplicateType replicateType = ReplicateType.DATA_LOSS; - LedgerFragment(LedgerHandle lh, + public LedgerFragment(LedgerHandle lh, long firstEntryId, long lastKnownEntryId, Set bookieIndexes) { @@ -56,7 +57,7 @@ public class LedgerFragment { || !ensemble.equals(ensembles.get(ensembles.lastKey())); } - LedgerFragment(LedgerFragment lf, Set subset) { + public LedgerFragment(LedgerFragment lf, Set subset) { this.ledgerId = lf.ledgerId; this.firstEntryId = lf.firstEntryId; this.lastKnownEntryId = lf.lastKnownEntryId; @@ -91,7 +92,7 @@ public boolean isClosed() { return isLedgerClosed; } - long getLedgerId() { + public long getLedgerId() { return ledgerId; } @@ -217,6 +218,14 @@ public List getEnsemble() { return this.ensemble; } + public ReplicateType getReplicateType() { + return replicateType; + } + + public void setReplicateType(ReplicateType replicateType) { + this.replicateType = replicateType; + } + @Override public String toString() { return String.format("Fragment(LedgerID: %d, FirstEntryID: %d[%d], " @@ -224,4 +233,12 @@ public String toString() { getFirstStoredEntryId(), lastKnownEntryId, getLastStoredEntryId(), getAddresses(), isLedgerClosed); } + + /** + * ReplicateType. + */ + public enum ReplicateType { + DATA_LOSS, + DATA_NOT_ADHERING_PLACEMENT + } } 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 653b1a5e5db..72858f188f7 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 @@ -235,6 +235,28 @@ public BookieNode selectFromNetworkLocation( } } + @Override + public PlacementResult> replaceToAdherePlacementPolicy( + int ensembleSize, + int writeQuorumSize, + int ackQuorumSize, + Set excludeBookies, + List currentEnsemble) { + final PlacementResult> placementResult = + super.replaceToAdherePlacementPolicy(ensembleSize, writeQuorumSize, ackQuorumSize, + excludeBookies, currentEnsemble); + if (placementResult.getAdheringToPolicy() != PlacementPolicyAdherence.FAIL) { + return placementResult; + } else { + if (slave == null) { + return placementResult; + } else { + return slave.replaceToAdherePlacementPolicy(ensembleSize, writeQuorumSize, ackQuorumSize, + excludeBookies, currentEnsemble); + } + } + } + @Override public void handleBookiesThatLeft(Set leftBookies) { super.handleBookiesThatLeft(leftBookies); 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 c6625b55e80..d3d557d14be 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 @@ -26,6 +26,7 @@ import static org.apache.bookkeeper.client.BookKeeperClientStats.READ_REQUESTS_REORDERED; import static org.apache.bookkeeper.client.RegionAwareEnsemblePlacementPolicy.UNKNOWN_REGION; +import com.beust.jcommander.internal.Lists; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; @@ -43,6 +44,8 @@ import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException; import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo; import org.apache.bookkeeper.client.WeightedRandomSelection.WeightedObject; @@ -66,6 +69,7 @@ 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; @@ -1079,4 +1083,221 @@ public boolean areAckedBookiesAdheringToPlacementPolicy(Set ackedBooki } return rackCounter.size() >= minWriteQuorumNumRacksPerWriteQuorum; } + + @Override + public PlacementResult> replaceToAdherePlacementPolicy( + int ensembleSize, + int writeQuorumSize, + int ackQuorumSize, + Set excludeBookies, + List currentEnsemble) { + rwLock.readLock().lock(); + try { + PlacementPolicyAdherence currentPlacementAdherence = isEnsembleAdheringToPlacementPolicy( + currentEnsemble, writeQuorumSize, ackQuorumSize); + if (PlacementPolicyAdherence.FAIL != currentPlacementAdherence) { + return PlacementResult.of(new ArrayList<>(currentEnsemble), currentPlacementAdherence); + } + for (BookieId bookieId : currentEnsemble) { + if (!knownBookies.containsKey(bookieId)) { + excludeBookies.add(bookieId); + } + } + int minNumRacksPerWriteQuorumForThisEnsemble = Math.min(writeQuorumSize, minNumRacksPerWriteQuorum); + int numRacks = topology.getNumOfRacks(); + // only one rack or less than minNumRacksPerWriteQuorumForThisEnsemble, stop calculation to skip relocation + if (numRacks < 2 || numRacks < minNumRacksPerWriteQuorumForThisEnsemble) { + LOG.warn("Skip ensemble relocation because the cluster has only {} rack.", numRacks); + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); + } + PlacementResult> placementResult = PlacementResult.of(Collections.emptyList(), + PlacementPolicyAdherence.FAIL); + int minDiffer = Integer.MAX_VALUE; + for (int i = 0; i < currentEnsemble.size(); i++) { + PlacementResult> result = doReplaceToAdherePlacementPolicy(ensembleSize, + writeQuorumSize, ackQuorumSize, excludeBookies, currentEnsemble, i); + if (PlacementPolicyAdherence.FAIL == result.getAdheringToPolicy()) { + continue; + } + int differ = differBetweenBookies(currentEnsemble, result.getResult()); + if (differ < minDiffer) { + minDiffer = differ; + placementResult = result; + if (minDiffer == 1) { + break; + } + } + } + return placementResult; + } finally { + rwLock.readLock().unlock(); + } + } + + private PlacementResult> doReplaceToAdherePlacementPolicy( + int ensembleSize, + int writeQuorumSize, + int ackQuorumSize, + Set excludeBookies, + List currentEnsemble, + int startIndex) { + final List provisionalEnsembleNodes = currentEnsemble.stream() + .map(this::convertBookieToNode).collect(Collectors.toList()); + final Set excludeNodes = convertBookiesToNodes( + addDefaultRackBookiesIfMinNumRacksIsEnforced(excludeBookies)); + int minNumRacksPerWriteQuorumForThisEnsemble = Math.min(writeQuorumSize, minNumRacksPerWriteQuorum); + final RRTopologyAwareCoverageEnsemble ensemble = + new RRTopologyAwareCoverageEnsemble( + ensembleSize, + writeQuorumSize, + ackQuorumSize, + RACKNAME_DISTANCE_FROM_LEAVES, + null, + null, + minNumRacksPerWriteQuorumForThisEnsemble); + BookieNode prevNode = null; + final BookieNode firstNode = provisionalEnsembleNodes.get(startIndex); + // use same bookie at first to reduce ledger replication + if (!excludeNodes.contains(firstNode) && ensemble.apply(firstNode, ensemble) + && ensemble.addNode(firstNode)) { + excludeNodes.add(firstNode); + prevNode = firstNode; + } + for (int i = prevNode == null ? 0 : 1; i < ensembleSize; i++) { + int index = (startIndex + i) % ensembleSize; + final String curRack; + if (null == prevNode) { + if ((null == localNode) || defaultRack.equals(localNode.getNetworkLocation())) { + curRack = NodeBase.ROOT; + } else { + curRack = localNode.getNetworkLocation(); + } + } else { + curRack = "~" + prevNode.getNetworkLocation(); + } + try { + prevNode = replaceToAdherePlacementPolicyInternal( + curRack, excludeNodes, ensemble, ensemble, + provisionalEnsembleNodes, index, ensembleSize, minNumRacksPerWriteQuorumForThisEnsemble); + // got a good candidate + if (ensemble.addNode(prevNode)) { + // add the candidate to exclude set + excludeNodes.add(prevNode); + } else { + throw new BKNotEnoughBookiesException(); + } + // replace to newer node + provisionalEnsembleNodes.set(index, prevNode); + } catch (BKNotEnoughBookiesException e) { + LOG.warn("Skip ensemble relocation because the cluster has not enough bookies."); + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); + } + } + List bookieList = ensemble.toList(); + if (ensembleSize != bookieList.size()) { + LOG.warn("Not enough {} bookies are available to form an ensemble : {}.", + ensembleSize, bookieList); + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); + } + PlacementPolicyAdherence placementPolicyAdherence = isEnsembleAdheringToPlacementPolicy(bookieList, + writeQuorumSize, ackQuorumSize); + if (PlacementPolicyAdherence.FAIL == placementPolicyAdherence) { + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); + } + return PlacementResult.of(revertBookieListByIndex(bookieList, startIndex), placementPolicyAdherence); + } + + private List revertBookieListByIndex(List bookies, int startIndex) { + BookieId[] bookieIds = new BookieId[bookies.size()]; + for (int i = 0; i < bookies.size(); i++) { + if (startIndex == bookies.size()) { + startIndex = 0; + } + bookieIds[startIndex++] = bookies.get(i); + } + return Lists.newArrayList(bookieIds); + } + + private BookieNode replaceToAdherePlacementPolicyInternal( + String netPath, Set excludeBookies, Predicate predicate, + Ensemble ensemble, List provisionalEnsembleNodes, int ensembleIndex, + int ensembleSize, int minNumRacksPerWriteQuorumForThisEnsemble) throws BKNotEnoughBookiesException { + final BookieNode currentNode = provisionalEnsembleNodes.get(ensembleIndex); + // if the current bookie could be applied to the ensemble, apply it to minify the number of bookies replaced + if (!excludeBookies.contains(currentNode) && predicate.apply(currentNode, ensemble)) { + return currentNode; + } + final List>> conditionList = new ArrayList<>(); + final Set preExcludeRacks = new HashSet<>(); + final Set postExcludeRacks = new HashSet<>(); + for (int i = 0; i < minNumRacksPerWriteQuorumForThisEnsemble - 1; i++) { + preExcludeRacks.add(provisionalEnsembleNodes.get(Math.floorMod((ensembleIndex - i - 1), ensembleSize)) + .getNetworkLocation()); + postExcludeRacks.add(provisionalEnsembleNodes.get(Math.floorMod((ensembleIndex + i + 1), ensembleSize)) + .getNetworkLocation()); + } + // adhere minNumRacksPerWriteQuorum by preExcludeRacks + // avoid additional replace from write quorum candidates by preExcludeRacks and postExcludeRacks + // avoid to use first candidate bookies for election by provisionalEnsembleNodes + conditionList.add(Pair.of( + "~" + String.join(",", + Stream.concat(preExcludeRacks.stream(), postExcludeRacks.stream()).collect(Collectors.toSet())), + provisionalEnsembleNodes + )); + // avoid to use same rack between previous index by netPath + // avoid to use first candidate bookies for election by provisionalEnsembleNodes + conditionList.add(Pair.of(netPath, provisionalEnsembleNodes)); + // avoid to use same rack between previous index by netPath + conditionList.add(Pair.of(netPath, Collections.emptyList())); + + for (Pair> condition : conditionList) { + WeightedRandomSelection wRSelection = null; + + final List leaves = new ArrayList<>(topology.getLeaves(condition.getLeft())); + if (!isWeighted) { + Collections.shuffle(leaves); + } else { + if (CollectionUtils.subtract(leaves, excludeBookies).size() < 1) { + throw new BKNotEnoughBookiesException(); + } + wRSelection = prepareForWeightedSelection(leaves); + if (wRSelection == null) { + throw new BKNotEnoughBookiesException(); + } + } + + final Iterator it = leaves.iterator(); + final Set bookiesSeenSoFar = new HashSet<>(); + while (true) { + Node n; + if (isWeighted) { + if (bookiesSeenSoFar.size() == leaves.size()) { + // Don't loop infinitely. + break; + } + n = wRSelection.getNextRandom(); + bookiesSeenSoFar.add(n); + } else { + if (it.hasNext()) { + n = it.next(); + } else { + break; + } + } + if (excludeBookies.contains(n)) { + continue; + } + if (!(n instanceof BookieNode) || !predicate.apply((BookieNode) n, ensemble)) { + continue; + } + // additional excludeBookies + if (condition.getRight().contains(n)) { + continue; + } + BookieNode bn = (BookieNode) n; + return bn; + } + } + throw new BKNotEnoughBookiesException(); + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TopologyAwareEnsemblePlacementPolicy.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TopologyAwareEnsemblePlacementPolicy.java index 805e60b1ac6..852b7f3e35d 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TopologyAwareEnsemblePlacementPolicy.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/TopologyAwareEnsemblePlacementPolicy.java @@ -52,6 +52,7 @@ import org.apache.bookkeeper.stats.Counter; import org.apache.bookkeeper.stats.OpStatsLogger; import org.apache.bookkeeper.stats.annotations.StatsDoc; +import org.apache.commons.collections4.CollectionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -763,6 +764,22 @@ public void onBookieRackChange(List bookieAddressList) { } } + public static int differBetweenBookies(List bookiesA, List bookiesB) { + if (CollectionUtils.isEmpty(bookiesA) || CollectionUtils.isEmpty(bookiesB)) { + return Integer.MAX_VALUE; + } + if (bookiesA.size() != bookiesB.size()) { + return Integer.MAX_VALUE; + } + int differ = 0; + for (int i = 0; i < bookiesA.size(); i++) { + if (!bookiesA.get(i).equals(bookiesB.get(i))) { + differ++; + } + } + return differ; + } + @Override public void updateBookieInfo(Map bookieInfoMap) { if (!isWeighted) { @@ -812,15 +829,19 @@ protected String resolveNetworkLocation(BookieId addr) { } } - protected Set convertBookiesToNodes(Collection excludeBookies) { + protected Set convertBookiesToNodes(Collection bookies) { Set nodes = new HashSet(); - for (BookieId addr : excludeBookies) { - BookieNode bn = knownBookies.get(addr); - if (null == bn) { - bn = createBookieNode(addr); - } - nodes.add(bn); + for (BookieId addr : bookies) { + nodes.add(convertBookieToNode(addr)); } return nodes; } + + protected BookieNode convertBookieToNode(BookieId addr) { + BookieNode bn = knownBookies.get(addr); + if (null == bn) { + bn = createBookieNode(addr); + } + return bn; + } } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java index eebda935595..1e441609660 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java @@ -202,6 +202,8 @@ public class ServerConfiguration extends AbstractConfiguration { + if (e != null) { + LOG.error("For ledger: {}, the placement policy not adhering bookie " + + "storage, mark it to under replication manager failed.", + ledgerId, e); + return; + } + if (LOG.isDebugEnabled()) { + LOG.debug("For ledger: {}, the placement policy not adhering bookie" + + " storage, mark it to under replication manager", ledgerId); + } + }); + } } else if (foundSegmentSoftlyAdheringToPlacementPolicy) { numOfLedgersFoundSoftlyAdheringInPlacementPolicyCheck.incrementAndGet(); } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java index c4a0e430d6f..e15f5c1407e 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/replication/ReplicationWorker.java @@ -34,7 +34,9 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.IOException; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.NavigableSet; @@ -42,6 +44,7 @@ import java.util.SortedMap; import java.util.Timer; import java.util.TimerTask; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -53,12 +56,15 @@ import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.BookKeeperAdmin; +import org.apache.bookkeeper.client.EnsemblePlacementPolicy; import org.apache.bookkeeper.client.LedgerChecker; import org.apache.bookkeeper.client.LedgerFragment; import org.apache.bookkeeper.client.LedgerHandle; import org.apache.bookkeeper.client.api.LedgerMetadata; +import org.apache.bookkeeper.common.concurrent.FutureUtils; import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.meta.LedgerManager; import org.apache.bookkeeper.meta.LedgerUnderreplicationManager; import org.apache.bookkeeper.net.BookieId; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback; @@ -69,6 +75,7 @@ import org.apache.bookkeeper.stats.OpStatsLogger; import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.stats.annotations.StatsDoc; +import org.apache.bookkeeper.versioning.Versioned; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -100,6 +107,7 @@ public class ReplicationWorker implements Runnable { private final long lockReleaseOfFailedLedgerGracePeriod; private final long baseBackoffForLockReleaseOfFailedLedger; private final BiConsumer onReadEntryFailureCallback; + private final LedgerManager ledgerManager; // Expose Stats private final StatsLogger statsLogger; @@ -173,6 +181,7 @@ public ReplicationWorker(final ServerConfiguration conf, this.ownBkc = ownBkc; this.underreplicationManager = bkc.getLedgerManagerFactory().newLedgerUnderreplicationManager(); + this.ledgerManager = bkc.getLedgerManagerFactory().newLedgerManager(); this.admin = new BookKeeperAdmin(bkc, statsLogger, new ClientConfiguration(conf)); this.ledgerChecker = new LedgerChecker(bkc); this.workerThread = new BookieThread(this, "ReplicationWorker"); @@ -359,6 +368,65 @@ private boolean tryReadingFaultyEntries(LedgerHandle lh, LedgerFragment ledgerFr return (returnRCValue.get() == BKException.Code.OK); } + private Set getNeedRepairedPlacementNotAdheringFragments(LedgerHandle lh) { + if (!conf.isRepairedPlacementPolicyNotAdheringBookieEnable()) { + return Collections.emptySet(); + } + long ledgerId = lh.getId(); + Set placementNotAdheringFragments = new HashSet<>(); + CompletableFuture> future = ledgerManager.readLedgerMetadata( + ledgerId).whenComplete((metadataVer, exception) -> { + if (exception == null) { + LedgerMetadata metadata = metadataVer.getValue(); + int writeQuorumSize = metadata.getWriteQuorumSize(); + int ackQuorumSize = metadata.getAckQuorumSize(); + if (!metadata.isClosed()) { + return; + } + Long curEntryId = null; + EnsemblePlacementPolicy.PlacementPolicyAdherence previousSegmentAdheringToPlacementPolicy = null; + + for (Map.Entry> entry : metadata.getAllEnsembles().entrySet()) { + if (curEntryId != null) { + if (EnsemblePlacementPolicy.PlacementPolicyAdherence.FAIL + == previousSegmentAdheringToPlacementPolicy) { + LedgerFragment ledgerFragment = new LedgerFragment(lh, curEntryId, + entry.getKey() - 1, new HashSet<>()); + ledgerFragment.setReplicateType(LedgerFragment.ReplicateType.DATA_NOT_ADHERING_PLACEMENT); + placementNotAdheringFragments.add(ledgerFragment); + } + } + previousSegmentAdheringToPlacementPolicy = + admin.isEnsembleAdheringToPlacementPolicy(entry.getValue(), + writeQuorumSize, ackQuorumSize); + curEntryId = entry.getKey(); + } + if (curEntryId != null) { + if (EnsemblePlacementPolicy.PlacementPolicyAdherence.FAIL + == previousSegmentAdheringToPlacementPolicy) { + long lastEntry = lh.getLedgerMetadata().getLastEntryId(); + LedgerFragment ledgerFragment = new LedgerFragment(lh, curEntryId, lastEntry, + new HashSet<>()); + ledgerFragment.setReplicateType(LedgerFragment.ReplicateType.DATA_NOT_ADHERING_PLACEMENT); + placementNotAdheringFragments.add(ledgerFragment); + } + } + } else if (BKException.getExceptionCode(exception) + == BKException.Code.NoSuchLedgerExistsOnMetadataServerException) { + LOG.debug("Ignoring replication of already deleted ledger {}", ledgerId); + } else { + LOG.warn("Unable to read the ledger: {} information", ledgerId); + } + }); + try { + FutureUtils.result(future); + } catch (Exception e) { + LOG.warn("Check ledger need repaired placement not adhering bookie failed", e); + return Collections.emptySet(); + } + return placementNotAdheringFragments; + } + @SuppressFBWarnings("RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE") private boolean rereplicate(long ledgerIdToReplicate) throws InterruptedException, BKException, UnavailableException { @@ -369,8 +437,8 @@ private boolean rereplicate(long ledgerIdToReplicate) throws InterruptedExceptio boolean deferLedgerLockRelease = false; try (LedgerHandle lh = admin.openLedgerNoRecovery(ledgerIdToReplicate)) { - Set fragments = - getUnderreplicatedFragments(lh, conf.getAuditorLedgerVerificationPercentage()); + Set fragments = getUnderreplicatedFragments(lh, + conf.getAuditorLedgerVerificationPercentage()); if (LOG.isDebugEnabled()) { LOG.debug("Founds fragments {} for replication from ledger: {}", fragments, ledgerIdToReplicate); @@ -502,10 +570,40 @@ private boolean isLastSegmentOpenAndMissingBookies(LedgerHandle lh) throws BKExc */ private Set getUnderreplicatedFragments(LedgerHandle lh, Long ledgerVerificationPercentage) throws InterruptedException { + //The data loss fragments is first to repair. If a fragment is data_loss and not_adhering_placement + //at the same time, we only fix data_loss in this time. After fix data_loss, the fragment is still + //not_adhering_placement, Auditor will mark this ledger again. + Set underreplicatedFragments = new HashSet<>(); + + Set dataLossFragments = getDataLossFragments(lh, ledgerVerificationPercentage); + underreplicatedFragments.addAll(dataLossFragments); + + Set notAdheringFragments = getNeedRepairedPlacementNotAdheringFragments(lh); + + for (LedgerFragment notAdheringFragment : notAdheringFragments) { + if (!checkFragmentRepeat(underreplicatedFragments, notAdheringFragment)) { + underreplicatedFragments.add(notAdheringFragment); + } + } + return underreplicatedFragments; + } + + private Set getDataLossFragments(LedgerHandle lh, Long ledgerVerificationPercentage) + throws InterruptedException { CheckerCallback checkerCb = new CheckerCallback(); ledgerChecker.checkLedger(lh, checkerCb, ledgerVerificationPercentage); - Set fragments = checkerCb.waitAndGetResult(); - return fragments; + return checkerCb.waitAndGetResult(); + } + + private boolean checkFragmentRepeat(Set fragments, LedgerFragment needChecked) { + for (LedgerFragment fragment : fragments) { + if (fragment.getLedgerId() == needChecked.getLedgerId() + && fragment.getFirstEntryId() == needChecked.getFirstEntryId() + && fragment.getLastKnownEntryId() == needChecked.getLastKnownEntryId()) { + return true; + } + } + return false; } void scheduleTaskWithDelay(TimerTask timerTask, long delayPeriod) { 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 0e9233136e2..396e86d92bb 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 @@ -21,6 +21,9 @@ import static org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicyImpl.shuffleWithMask; import static org.apache.bookkeeper.client.RoundRobinDistributionSchedule.writeSetFromValues; import static org.apache.bookkeeper.feature.SettableFeatureProvider.DISABLE_ALL; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertThat; import com.google.common.util.concurrent.ThreadFactoryBuilder; import io.netty.util.HashedWheelTimer; @@ -36,6 +39,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import junit.framework.TestCase; import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException; import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo; @@ -60,6 +64,10 @@ import org.apache.bookkeeper.test.TestStatsProvider.TestStatsLogger; import org.apache.bookkeeper.util.StaticDNSResolver; import org.apache.commons.collections4.CollectionUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.hamcrest.Description; +import org.hamcrest.Matcher; +import org.hamcrest.TypeSafeMatcher; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -650,7 +658,7 @@ public void testReplaceBookieWithEnoughBookiesInSameRack() throws Exception { EnsemblePlacementPolicy.PlacementResult replaceBookieResponse = repp.replaceBookie(1, 1, 1, null, new ArrayList<>(), addr2.toBookieId(), new HashSet<>()); BookieId replacedBookie = replaceBookieResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertEquals(addr3.toBookieId(), replacedBookie); assertEquals(PlacementPolicyAdherence.MEETS_STRICT, isEnsembleAdheringToPlacementPolicy); } @@ -679,7 +687,7 @@ public void testReplaceBookieWithEnoughBookiesInDifferentRack() throws Exception EnsemblePlacementPolicy.PlacementResult replaceBookieResponse = repp.replaceBookie(1, 1, 1, null, new ArrayList<>(), addr2.toBookieId(), excludedAddrs); BookieId replacedBookie = replaceBookieResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertFalse(addr1.toBookieId().equals(replacedBookie)); assertTrue(addr3.toBookieId().equals(replacedBookie) || addr4.toBookieId().equals(replacedBookie)); assertEquals(PlacementPolicyAdherence.MEETS_STRICT, isEnsembleAdheringToPlacementPolicy); @@ -744,7 +752,7 @@ public void testReplaceBookieWithEnoughBookiesInSameRackAsEnsemble() throws Exce addr4.toBookieId(), new HashSet<>()); BookieId replacedBookie = replaceBookieResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertEquals(addr1.toBookieId(), replacedBookie); assertEquals(PlacementPolicyAdherence.MEETS_STRICT, isEnsembleAdheringToPlacementPolicy); } @@ -766,14 +774,14 @@ public void testNewEnsembleWithSingleRack() throws Exception { EnsemblePlacementPolicy.PlacementResult> ensembleResponse; ensembleResponse = repp.newEnsemble(3, 2, 2, null, new HashSet<>()); List ensemble = ensembleResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals(0, getNumCoveredWriteQuorums(ensemble, 2, conf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); assertEquals(PlacementPolicyAdherence.FAIL, isEnsembleAdheringToPlacementPolicy); EnsemblePlacementPolicy.PlacementResult> ensembleResponse2; ensembleResponse2 = repp.newEnsemble(4, 2, 2, null, new HashSet<>()); List ensemble2 = ensembleResponse2.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy2 = ensembleResponse2.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy2 = ensembleResponse2.getAdheringToPolicy(); assertEquals(0, getNumCoveredWriteQuorums(ensemble2, 2, conf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); assertEquals(PlacementPolicyAdherence.FAIL, isEnsembleAdheringToPlacementPolicy); @@ -898,7 +906,7 @@ public void testNewEnsembleWithEnforceMinNumRacks() throws Exception { ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, null, new HashSet<>()); ensemble = ensembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals("Number of writeQuorum sets covered", ensembleSize, getNumCoveredWriteQuorums(ensemble, writeQuorumSize, clientConf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -908,7 +916,7 @@ public void testNewEnsembleWithEnforceMinNumRacks() throws Exception { new HashSet<>(defaultRackBookiesList), EnsembleForReplacementWithNoConstraints.INSTANCE, TruePredicate.INSTANCE); ensemble = ensembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals("Number of writeQuorum sets covered", ensembleSize, getNumCoveredWriteQuorums(ensemble, writeQuorumSize, clientConf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -962,7 +970,7 @@ public void testNewEnsembleWithSufficientRacksAndEnforceMinNumRacks() throws Exc for (int ensembleSize = effectiveMinNumRacksPerWriteQuorum; ensembleSize < 40; ensembleSize++) { ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, null, new HashSet<>()); ensemble = ensembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals("Number of writeQuorum sets covered", ensembleSize, getNumCoveredWriteQuorums(ensemble, writeQuorumSize, clientConf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -971,7 +979,7 @@ public void testNewEnsembleWithSufficientRacksAndEnforceMinNumRacks() throws Exc ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, new HashSet<>(), EnsembleForReplacementWithNoConstraints.INSTANCE, TruePredicate.INSTANCE); ensemble = ensembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals("Number of writeQuorum sets covered", ensembleSize, getNumCoveredWriteQuorums(ensemble, writeQuorumSize, clientConf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -1098,7 +1106,7 @@ public void testReplaceBookieWithEnforceMinNumRacks() throws Exception { replaceBookieResponse = repp.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize, null, ensemble, bookieInEnsembleToBeReplaced, new HashSet<>()); replacedBookieAddress = replaceBookieResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertEquals("It should be newBookieAddress2", newBookieAddress2.toBookieId(), replacedBookieAddress); assertEquals(PlacementPolicyAdherence.MEETS_STRICT, isEnsembleAdheringToPlacementPolicy); @@ -1138,7 +1146,7 @@ public void testReplaceBookieWithEnforceMinNumRacks() throws Exception { replaceBookieResponse = repp.replaceBookie(ensembleSize, writeQuorumSize, ackQuorumSize, null, ensemble, bookieInEnsembleToBeReplaced, bookiesToExclude); replacedBookieAddress = replaceBookieResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertEquals("It should be newBookieAddress3", newBookieAddress3.toBookieId(), replacedBookieAddress); assertEquals(PlacementPolicyAdherence.MEETS_STRICT, isEnsembleAdheringToPlacementPolicy); } @@ -1491,7 +1499,7 @@ public void testNewEnsembleWithMultipleRacks() throws Exception { repp.newEnsemble(ensembleSize, writeQuorumSize, acqQuorumSize, null, new HashSet<>()); List ensemble = ensembleResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); int numCovered = getNumCoveredWriteQuorums(ensemble, writeQuorumSize, conf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver); assertTrue(numCovered >= 1 && numCovered < 3); @@ -1501,7 +1509,7 @@ public void testNewEnsembleWithMultipleRacks() throws Exception { repp.newEnsemble(ensembleSize, writeQuorumSize, acqQuorumSize, null, new HashSet<>()); List ensemble2 = ensembleResponse2.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy2 = ensembleResponse2.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy2 = ensembleResponse2.getAdheringToPolicy(); numCovered = getNumCoveredWriteQuorums(ensemble2, writeQuorumSize, conf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver); assertTrue(numCovered >= 1 && numCovered < 3); @@ -1579,7 +1587,7 @@ void validateNumOfWriteQuorumsCoveredInEnsembleCreation(Set addrs, repp.newEnsemble(ensembleSize, writeQuorumSize, writeQuorumSize, null, new HashSet<>()); List ensemble = ensembleResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); int numCovered = getNumCoveredWriteQuorums(ensemble, writeQuorumSize, minNumRacksPerWriteQuorum, repp.bookieAddressResolver); assertEquals("minimum number of racks covered for writequorum ensemble: " + ensemble, ensembleSize, numCovered); @@ -1625,7 +1633,7 @@ public void testNewEnsembleWithEnoughRacks() throws Exception { repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, null, new HashSet<>()); List ensemble1 = ensembleResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy1 = ensembleResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy1 = ensembleResponse.getAdheringToPolicy(); assertEquals(ensembleSize, getNumCoveredWriteQuorums(ensemble1, writeQuorumSize, conf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -1635,7 +1643,7 @@ public void testNewEnsembleWithEnoughRacks() throws Exception { EnsemblePlacementPolicy.PlacementResult> ensembleResponse2 = repp.newEnsemble(ensembleSize, writeQuorumSize, 2, null, new HashSet<>()); List ensemble2 = ensembleResponse2.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy2 = ensembleResponse2.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy2 = ensembleResponse2.getAdheringToPolicy(); assertEquals(ensembleSize, getNumCoveredWriteQuorums(ensemble2, writeQuorumSize, conf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -1719,7 +1727,7 @@ public void testWeightedPlacementAndReplaceBookieWithEnoughBookiesInSameRack() t replaceBookieResponse = repp.replaceBookie(1, 1, 1, null, new ArrayList<>(), addr2.toBookieId(), new HashSet<>()); replacedBookie = replaceBookieResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertTrue("replaced : " + replacedBookie, addr3.toBookieId().equals(replacedBookie) || addr4.toBookieId().equals(replacedBookie)); assertEquals(PlacementPolicyAdherence.MEETS_STRICT, isEnsembleAdheringToPlacementPolicy); @@ -1790,7 +1798,7 @@ public void testWeightedPlacementAndReplaceBookieWithoutEnoughBookiesInSameRack( replaceBookieResponse = repp.replaceBookie(1, 1, 1, null, new ArrayList<>(), addr2.toBookieId(), new HashSet<>()); replacedBookie = replaceBookieResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = replaceBookieResponse.getAdheringToPolicy(); assertTrue(addr0.toBookieId().equals(replacedBookie) || addr1.toBookieId().equals(replacedBookie) || addr3.toBookieId().equals(replacedBookie) @@ -2072,7 +2080,7 @@ public void testPlacementOnStabilizeNetworkTopology() throws Exception { EnsemblePlacementPolicy.PlacementResult> ensembleResponse = repp.newEnsemble(3, 2, 2, null, new HashSet()); List ensemble = ensembleResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertFalse(ensemble.contains(addr4.toBookieId())); assertEquals(PlacementPolicyAdherence.FAIL, isEnsembleAdheringToPlacementPolicy); } @@ -2081,7 +2089,7 @@ public void testPlacementOnStabilizeNetworkTopology() throws Exception { EnsemblePlacementPolicy.PlacementResult> ensembleResponse = repp.newEnsemble(4, 2, 2, null, new HashSet()); List ensemble = ensembleResponse.getResult(); - PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + PlacementPolicyAdherence isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals(PlacementPolicyAdherence.FAIL, isEnsembleAdheringToPlacementPolicy); assertTrue(ensemble.contains(addr4.toBookieId())); } @@ -2375,7 +2383,7 @@ public void testNewEnsembleExcludesDefaultRackBookiesEnforceMinNumRacks() throws for (int ensembleSize = effectiveMinNumRacksPerWriteQuorum; ensembleSize < 40; ensembleSize++) { ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, null, new HashSet<>()); ensemble = ensembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals("Number of writeQuorum sets covered", ensembleSize, getNumCoveredWriteQuorums(ensemble, writeQuorumSize, clientConf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -2383,7 +2391,7 @@ public void testNewEnsembleExcludesDefaultRackBookiesEnforceMinNumRacks() throws ensembleResponse = repp.newEnsemble(ensembleSize, writeQuorumSize, ackQuorumSize, null, new HashSet<>()); ensemble = ensembleResponse.getResult(); - isEnsembleAdheringToPlacementPolicy = ensembleResponse.isAdheringToPolicy(); + isEnsembleAdheringToPlacementPolicy = ensembleResponse.getAdheringToPolicy(); assertEquals("Number of writeQuorum sets covered", ensembleSize, getNumCoveredWriteQuorums(ensemble, writeQuorumSize, clientConf.getMinNumRacksPerWriteQuorum(), repp.bookieAddressResolver)); @@ -2484,4 +2492,384 @@ public void testAreAckedBookiesAdheringToPlacementPolicy() throws Exception { testAreAckedBookiesAdheringToPlacementPolicyHelper(4, 6, 3, 2, 6, 3, 3); testAreAckedBookiesAdheringToPlacementPolicyHelper(5, 7, 5, 3, 7, 5, 2); } + + @SuppressWarnings("unchecked") + @Test + public void testReplaceToAdherePlacementPolicy() throws Exception { + final BookieSocketAddress addr1 = new BookieSocketAddress("127.0.0.1", 3181); + final BookieSocketAddress addr2 = new BookieSocketAddress("127.0.0.2", 3181); + final BookieSocketAddress addr3 = new BookieSocketAddress("127.0.0.3", 3181); + final BookieSocketAddress addr4 = new BookieSocketAddress("127.0.0.4", 3181); + final BookieSocketAddress addr5 = new BookieSocketAddress("127.0.0.5", 3181); + final BookieSocketAddress addr6 = new BookieSocketAddress("127.0.0.6", 3181); + final BookieSocketAddress addr7 = new BookieSocketAddress("127.0.0.7", 3181); + final BookieSocketAddress addr8 = new BookieSocketAddress("127.0.0.8", 3181); + final BookieSocketAddress addr9 = new BookieSocketAddress("127.0.0.9", 3181); + + final String rackName1 = NetworkTopology.DEFAULT_REGION + "/r1"; + final String rackName2 = NetworkTopology.DEFAULT_REGION + "/r2"; + final String rackName3 = NetworkTopology.DEFAULT_REGION + "/r3"; + + // update dns mapping + StaticDNSResolver.addNodeToRack(addr1.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr2.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr3.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr4.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr5.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr6.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr7.getSocketAddress().getAddress().getHostAddress(), rackName3); + StaticDNSResolver.addNodeToRack(addr8.getSocketAddress().getAddress().getHostAddress(), rackName3); + StaticDNSResolver.addNodeToRack(addr9.getSocketAddress().getAddress().getHostAddress(), rackName3); + + // Update cluster + final Set addrs = new HashSet<>(); + addrs.add(addr1.toBookieId()); + addrs.add(addr2.toBookieId()); + addrs.add(addr3.toBookieId()); + addrs.add(addr4.toBookieId()); + addrs.add(addr5.toBookieId()); + addrs.add(addr6.toBookieId()); + addrs.add(addr7.toBookieId()); + addrs.add(addr8.toBookieId()); + addrs.add(addr9.toBookieId()); + + final ClientConfiguration newConf = new ClientConfiguration(conf); + newConf.setDiskWeightBasedPlacementEnabled(false); + newConf.setMinNumRacksPerWriteQuorum(2); + newConf.setEnforceMinNumRacksPerWriteQuorum(true); + + repp.initialize(newConf, Optional.empty(), timer, + DISABLE_ALL, NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER); + repp.withDefaultRack(NetworkTopology.DEFAULT_REGION_AND_RACK); + + repp.onClusterChanged(addrs, new HashSet<>()); + final Map bookieInfoMap = new HashMap<>(); + bookieInfoMap.put(addr1.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr2.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr3.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr4.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr5.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr6.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr7.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr8.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr9.toBookieId(), new BookieInfo(100L, 100L)); + + repp.updateBookieInfo(bookieInfoMap); + + final Set excludeList = new HashSet<>(); + final int ensembleSize = 7; + final int writeQuorumSize = 2; + final int ackQuorumSize = 2; + + final BookieRackMatcher rack1 = new BookieRackMatcher(rackName1); + final BookieRackMatcher rack2 = new BookieRackMatcher(rackName2); + final BookieRackMatcher rack3 = new BookieRackMatcher(rackName3); + final BookieRackMatcher rack12 = new BookieRackMatcher(rackName1, rackName2); + final BookieRackMatcher rack13 = new BookieRackMatcher(rackName1, rackName3); + final BookieRackMatcher rack23 = new BookieRackMatcher(rackName2, rackName3); + final BookieRackMatcher rack123 = new BookieRackMatcher(rackName1, rackName2, rackName3); + final Consumer, Matcher>>> test = (pair) -> { + // RackawareEnsemblePlacementPolicyImpl#isEnsembleAdheringToPlacementPolicy + // is not scope of this test case. So, use the method in assertion for convenience. + assertEquals(PlacementPolicyAdherence.FAIL, + repp.isEnsembleAdheringToPlacementPolicy(pair.getLeft(), writeQuorumSize, ackQuorumSize)); + final EnsemblePlacementPolicy.PlacementResult> result = + repp.replaceToAdherePlacementPolicy(ensembleSize, writeQuorumSize, ackQuorumSize, + excludeList, pair.getLeft()); + if (LOG.isDebugEnabled()) { + LOG.debug("input: {}, result: {}", pair.getLeft(), result.getResult()); + } + assertEquals(PlacementPolicyAdherence.MEETS_STRICT, result.getAdheringToPolicy()); + assertThat(result.getResult(), pair.getRight()); + }; + + for (int i = 0; i < 1000; i++) { + test.accept(Pair.of(Arrays.asList(addr1.toBookieId(), addr4.toBookieId(), addr7.toBookieId(), + addr2.toBookieId(), addr5.toBookieId(), addr8.toBookieId(), addr9.toBookieId()), + // first, same, same, same, same, same, condition[0] + contains(is(addr1.toBookieId()), is(addr4.toBookieId()), is(addr7.toBookieId()), + is(addr2.toBookieId()), is(addr5.toBookieId()), is(addr8.toBookieId()), + is(addr6.toBookieId())))); + + test.accept(Pair.of(Arrays.asList(addr6.toBookieId(), addr4.toBookieId(), addr7.toBookieId(), + addr2.toBookieId(), addr5.toBookieId(), addr8.toBookieId(), addr3.toBookieId()), + // first, condition[0], same, same, same, same, same + contains(is(addr6.toBookieId()), is(addr1.toBookieId()), is(addr7.toBookieId()), + is(addr2.toBookieId()), is(addr5.toBookieId()), is(addr8.toBookieId()), + is(addr3.toBookieId())))); + + test.accept(Pair.of(Arrays.asList(addr1.toBookieId(), addr2.toBookieId(), addr3.toBookieId(), + addr4.toBookieId(), addr5.toBookieId(), addr6.toBookieId(), addr7.toBookieId()), + // first, candidate[0], same, same, candidate[0], same, same + contains(is(addr1.toBookieId()), is(rack3), is(addr3.toBookieId()), + is(addr4.toBookieId()), is(rack13), is(addr6.toBookieId()), is(addr7.toBookieId())))); + + test.accept(Pair.of(Arrays.asList(addr1.toBookieId(), addr2.toBookieId(), addr4.toBookieId(), + addr5.toBookieId(), addr7.toBookieId(), addr8.toBookieId(), addr9.toBookieId()), + contains(is(addr1.toBookieId()), is(rack23), is(rack123), is(rack123), + is(rack123), is(rack123), is(rack23)))); + } + StaticDNSResolver.reset(); + } + + @SuppressWarnings("unchecked") + @Test + public void testReplaceToAdherePlacementPolicyWithOutOfOrder() throws Exception { + final BookieSocketAddress addr1 = new BookieSocketAddress("127.0.0.1", 3181); + final BookieSocketAddress addr2 = new BookieSocketAddress("127.0.0.2", 3181); + final BookieSocketAddress addr3 = new BookieSocketAddress("127.0.0.3", 3181); + final BookieSocketAddress addr4 = new BookieSocketAddress("127.0.0.4", 3181); + final BookieSocketAddress addr5 = new BookieSocketAddress("127.0.0.5", 3181); + final BookieSocketAddress addr6 = new BookieSocketAddress("127.0.0.6", 3181); + + final String rackName1 = NetworkTopology.DEFAULT_REGION + "/r1"; + final String rackName2 = NetworkTopology.DEFAULT_REGION + "/r2"; + + // update dns mapping + StaticDNSResolver.addNodeToRack(addr1.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr2.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr3.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr4.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr5.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr6.getSocketAddress().getAddress().getHostAddress(), rackName2); + + // Update cluster + final Set addrs = new HashSet<>(); + addrs.add(addr1.toBookieId()); + addrs.add(addr2.toBookieId()); + addrs.add(addr3.toBookieId()); + addrs.add(addr4.toBookieId()); + addrs.add(addr5.toBookieId()); + addrs.add(addr6.toBookieId()); + + final ClientConfiguration newConf = new ClientConfiguration(conf); + newConf.setDiskWeightBasedPlacementEnabled(false); + newConf.setMinNumRacksPerWriteQuorum(2); + newConf.setEnforceMinNumRacksPerWriteQuorum(true); + + repp.initialize(newConf, Optional.empty(), timer, + DISABLE_ALL, NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER); + repp.withDefaultRack(NetworkTopology.DEFAULT_REGION_AND_RACK); + + repp.onClusterChanged(addrs, new HashSet<>()); + final Map bookieInfoMap = new HashMap<>(); + bookieInfoMap.put(addr1.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr2.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr3.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr4.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr5.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr6.toBookieId(), new BookieInfo(100L, 100L)); + + repp.updateBookieInfo(bookieInfoMap); + + final Set excludeList = new HashSet<>(); + final int ensembleSize = 6; + final int writeQuorumSize = 2; + final int ackQuorumSize = 2; + + final Consumer, Matcher>>> test = (pair) -> { + // RackawareEnsemblePlacementPolicyImpl#isEnsembleAdheringToPlacementPolicy + // is not scope of this test case. So, use the method in assertion for convenience. + assertEquals(PlacementPolicyAdherence.FAIL, + repp.isEnsembleAdheringToPlacementPolicy(pair.getLeft(), writeQuorumSize, ackQuorumSize)); + final EnsemblePlacementPolicy.PlacementResult> result = + repp.replaceToAdherePlacementPolicy(ensembleSize, writeQuorumSize, ackQuorumSize, + excludeList, pair.getLeft()); + if (LOG.isDebugEnabled()) { + LOG.debug("input: {}, result: {}", pair.getLeft(), result.getResult()); + } + assertEquals(PlacementPolicyAdherence.MEETS_STRICT, result.getAdheringToPolicy()); + }; + + for (int i = 0; i < 1000; i++) { + //All bookies already in the ensemble, the bookie order not adhere the placement policy. + test.accept(Pair.of(Arrays.asList(addr1.toBookieId(), addr2.toBookieId(), addr3.toBookieId(), + addr4.toBookieId(), addr5.toBookieId(), addr6.toBookieId()), + //The result is not predict. We know the best min replace place is 2. + //1,2,3,4,5,6 => 1,5,3,4,2,6 + //But maybe the final result is 1,6,3,4,2,5. + //When we from index 0 to replace, the first bookie(1) is /rack1, we only pick /rack2 bookie + //for the second bookie, so we can choose 4,5,6, the choice is random. If we pick 6 for the second, + // the final result is 1,6,3,4,2,5. If we pick 5 for the second, the final result is 1,5,3,4,2,6 + null)); + } + StaticDNSResolver.reset(); + } + + @SuppressWarnings("unchecked") + @Test + public void testReplaceToAdherePlacementPolicyWithNoMoreRackBookie() throws Exception { + final BookieSocketAddress addr1 = new BookieSocketAddress("127.0.0.1", 3181); + final BookieSocketAddress addr2 = new BookieSocketAddress("127.0.0.2", 3181); + final BookieSocketAddress addr3 = new BookieSocketAddress("127.0.0.3", 3181); + final BookieSocketAddress addr4 = new BookieSocketAddress("127.0.0.4", 3181); + final BookieSocketAddress addr5 = new BookieSocketAddress("127.0.0.5", 3181); + final BookieSocketAddress addr6 = new BookieSocketAddress("127.0.0.6", 3181); + + final String rackName1 = NetworkTopology.DEFAULT_REGION + "/r1"; + final String rackName2 = NetworkTopology.DEFAULT_REGION + "/r2"; + + // update dns mapping + StaticDNSResolver.addNodeToRack(addr1.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr2.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr3.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr4.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr5.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr6.getSocketAddress().getAddress().getHostAddress(), rackName2); + + // Update cluster + final Set addrs = new HashSet<>(); + addrs.add(addr1.toBookieId()); + addrs.add(addr2.toBookieId()); + addrs.add(addr3.toBookieId()); + addrs.add(addr4.toBookieId()); + addrs.add(addr5.toBookieId()); + addrs.add(addr6.toBookieId()); + + final ClientConfiguration newConf = new ClientConfiguration(conf); + newConf.setDiskWeightBasedPlacementEnabled(false); + newConf.setMinNumRacksPerWriteQuorum(2); + newConf.setEnforceMinNumRacksPerWriteQuorum(true); + + repp.initialize(newConf, Optional.empty(), timer, + DISABLE_ALL, NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER); + repp.withDefaultRack(NetworkTopology.DEFAULT_REGION_AND_RACK); + + repp.onClusterChanged(addrs, new HashSet<>()); + final Map bookieInfoMap = new HashMap<>(); + bookieInfoMap.put(addr1.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr2.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr3.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr4.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr5.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr6.toBookieId(), new BookieInfo(100L, 100L)); + + repp.updateBookieInfo(bookieInfoMap); + + final Set excludeList = new HashSet<>(); + final int ensembleSize = 3; + final int writeQuorumSize = 2; + final int ackQuorumSize = 2; + + final Consumer, Matcher>>> test = (pair) -> { + // RackawareEnsemblePlacementPolicyImpl#isEnsembleAdheringToPlacementPolicy + // is not scope of this test case. So, use the method in assertion for convenience. + assertEquals(PlacementPolicyAdherence.FAIL, + repp.isEnsembleAdheringToPlacementPolicy(pair.getLeft(), writeQuorumSize, ackQuorumSize)); + final EnsemblePlacementPolicy.PlacementResult> result = + repp.replaceToAdherePlacementPolicy(ensembleSize, writeQuorumSize, ackQuorumSize, + excludeList, pair.getLeft()); + if (LOG.isDebugEnabled()) { + LOG.debug("input: {}, result: {}", pair.getLeft(), result.getResult()); + } + assertEquals(PlacementPolicyAdherence.FAIL, result.getAdheringToPolicy()); + assertEquals(0, result.getResult().size()); + }; + + for (int i = 0; i < 1000; i++) { + test.accept(Pair.of(Arrays.asList(addr1.toBookieId(), addr2.toBookieId(), addr4.toBookieId()), + null)); + } + StaticDNSResolver.reset(); + } + + @SuppressWarnings("unchecked") + @Test + public void testReplaceToAdherePlacementPolicyWithUnknowBookie() throws Exception { + final BookieSocketAddress addr1 = new BookieSocketAddress("127.0.0.1", 3181); + final BookieSocketAddress addr2 = new BookieSocketAddress("127.0.0.2", 3181); + final BookieSocketAddress addr3 = new BookieSocketAddress("127.0.0.3", 3181); + final BookieSocketAddress addr4 = new BookieSocketAddress("127.0.0.4", 3181); + final BookieSocketAddress addr5 = new BookieSocketAddress("127.0.0.5", 3181); + final BookieSocketAddress addr6 = new BookieSocketAddress("127.0.0.6", 3181); + + final String rackName1 = NetworkTopology.DEFAULT_REGION + "/r1"; + final String rackName2 = NetworkTopology.DEFAULT_REGION + "/r2"; + + // update dns mapping + StaticDNSResolver.addNodeToRack(addr1.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr2.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr3.getSocketAddress().getAddress().getHostAddress(), rackName1); + StaticDNSResolver.addNodeToRack(addr4.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr5.getSocketAddress().getAddress().getHostAddress(), rackName2); + StaticDNSResolver.addNodeToRack(addr6.getSocketAddress().getAddress().getHostAddress(), rackName2); + + // Update cluster + final Set addrs = new HashSet<>(); + addrs.add(addr1.toBookieId()); + addrs.add(addr2.toBookieId()); + addrs.add(addr3.toBookieId()); + addrs.add(addr4.toBookieId()); + addrs.add(addr5.toBookieId()); + addrs.add(addr6.toBookieId()); + + final ClientConfiguration newConf = new ClientConfiguration(conf); + newConf.setDiskWeightBasedPlacementEnabled(false); + newConf.setMinNumRacksPerWriteQuorum(2); + newConf.setEnforceMinNumRacksPerWriteQuorum(true); + + repp.initialize(newConf, Optional.empty(), timer, + DISABLE_ALL, NullStatsLogger.INSTANCE, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER); + repp.withDefaultRack(NetworkTopology.DEFAULT_REGION_AND_RACK); + + repp.onClusterChanged(addrs, new HashSet<>()); + final Map bookieInfoMap = new HashMap<>(); + bookieInfoMap.put(addr1.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr2.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr3.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr4.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr5.toBookieId(), new BookieInfo(100L, 100L)); + bookieInfoMap.put(addr6.toBookieId(), new BookieInfo(100L, 100L)); + + repp.updateBookieInfo(bookieInfoMap); + + final Set excludeList = new HashSet<>(); + final int ensembleSize = 6; + final int writeQuorumSize = 2; + final int ackQuorumSize = 2; + + final BookieRackMatcher rack1 = new BookieRackMatcher(rackName1); + + final Consumer, Matcher>>> test = (pair) -> { + // RackawareEnsemblePlacementPolicyImpl#isEnsembleAdheringToPlacementPolicy + // is not scope of this test case. So, use the method in assertion for convenience. + assertEquals(PlacementPolicyAdherence.FAIL, + repp.isEnsembleAdheringToPlacementPolicy(pair.getLeft(), writeQuorumSize, ackQuorumSize)); + final EnsemblePlacementPolicy.PlacementResult> result = + repp.replaceToAdherePlacementPolicy(ensembleSize, writeQuorumSize, ackQuorumSize, + excludeList, pair.getLeft()); + if (LOG.isDebugEnabled()) { + LOG.debug("input: {}, result: {}", pair.getLeft(), result.getResult()); + } + assertEquals(PlacementPolicyAdherence.MEETS_STRICT, result.getAdheringToPolicy()); + assertThat(result.getResult(), pair.getRight()); + }; + + for (int i = 0; i < 1000; i++) { + test.accept(Pair.of(Arrays.asList(BookieId.parse("127.0.0.10:3181"), BookieId.parse("127.0.0.11:3181"), + addr3.toBookieId(), + addr4.toBookieId(), addr5.toBookieId(), addr6.toBookieId()), + contains(is(rack1), is(addr5.toBookieId()), is(addr3.toBookieId()), + is(addr4.toBookieId()), is(rack1), is(addr6.toBookieId())))); + } + StaticDNSResolver.reset(); + } + + private static class BookieRackMatcher extends TypeSafeMatcher { + final List expectedRacks; + + public BookieRackMatcher(String... expectedRacks) { + this.expectedRacks = Arrays.asList(expectedRacks); + } + + @Override + protected boolean matchesSafely(BookieId bookieId) { + return expectedRacks.contains(StaticDNSResolver.getRack(bookieId.toString().split(":")[0])); + } + + @Override + public void describeTo(Description description) { + description.appendText("expected racks " + expectedRacks); + } + } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestZoneawareEnsemblePlacementPolicy.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestZoneawareEnsemblePlacementPolicy.java index 1f7ebde9b0b..b191225c834 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestZoneawareEnsemblePlacementPolicy.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestZoneawareEnsemblePlacementPolicy.java @@ -241,7 +241,7 @@ public void testEnoughRWBookies() throws Exception { newEnsemblePlacementResult.getResult()); assertTrue("New ensemble should contain all 6 rw bookies", newEnsembleSet.containsAll(rwAddrs)); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); /* * there are enough bookies so newEnsemble should succeed. @@ -251,7 +251,7 @@ public void testEnoughRWBookies() throws Exception { assertTrue("New ensemble should contain 3 rw bookies", (newEnsembleSet.size() == 3) && (rwAddrs.containsAll(newEnsembleSet))); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); } @Test @@ -309,7 +309,7 @@ public void testWithDefaultBookies() throws Exception { assertTrue("Bookie from default faultDomain shouldn't be part of ensemble", Collections.disjoint(newEnsembleSet, bookiesInDefaultFaultDomain)); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); } } @@ -373,7 +373,7 @@ public void testMinZonesPerWriteQuorum() throws Exception { assertTrue("Bookie from default faultDomain shouldn't be part of ensemble", Collections.disjoint(newEnsembleSet, bookiesInDefaultFaultDomain)); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_SOFT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); try { /* @@ -404,7 +404,7 @@ public void testMinZonesPerWriteQuorum() throws Exception { newEnsembleSet = new HashSet(newEnsemblePlacementResult.getResult()); assertTrue("New ensemble should contain 4 different bookies", newEnsembleSet.size() == 4); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.FAIL, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); } @Test @@ -479,7 +479,7 @@ public void testMinUDsNotAvailable() throws Exception { */ newEnsemblePlacementResult = zepp.newEnsemble(ensSize, writeQuorum, 2, null, new HashSet<>()); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); List newEnsemble = newEnsemblePlacementResult.getResult(); Set newEnsembleSet = new HashSet(newEnsemble); assertTrue("New ensemble should contain all 6 rw bookies in non-default fault domains", @@ -563,7 +563,7 @@ public void testUniqueUds() throws Exception { assertTrue("New ensemble should contain 6 rw bookies in non-default fault domains", rwAddrs.containsAll(newEnsembleSet) && (newEnsembleSet.size() == 6)); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_SOFT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); Set bookiesNetworkLocations = new HashSet(); for (BookieId bookieAddr : newEnsembleSet) { @@ -654,7 +654,7 @@ public void testNewBookieUniformDistributionWithMinZoneAndMinUDs() throws Except excludedBookies); List newEnsembleList = newEnsemblePlacementResult.getResult(); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_SOFT, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); Set newEnsembleSet = new HashSet(newEnsembleList); Set bookiesNetworkLocationsSet = new HashSet(); List bookiesNodeLocationList = new ArrayList(); @@ -761,7 +761,7 @@ public void testReplaceBookie() throws Exception { BookieId replacedBookie = replacePlacementResult.getResult(); assertEquals("replaced bookie", addr8.toBookieId(), replacedBookie); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - replacePlacementResult.isAdheringToPolicy()); + replacePlacementResult.getAdheringToPolicy()); excludedBookies.add(addr8.toBookieId()); /* @@ -854,7 +854,7 @@ public void testReplaceBookieMinUDs() throws Exception { BookieId replacedBookie = replaceResponse.getResult(); assertEquals("replaced bookie", "/zone3/ud2", zepp.resolveNetworkLocation(replacedBookie)); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_SOFT, - replaceResponse.isAdheringToPolicy()); + replaceResponse.getAdheringToPolicy()); } @Test @@ -1074,7 +1074,7 @@ public void testPlacementOnStabilizeNetworkTopology() throws Exception { List ensemble = ensembleResponse.getResult(); assertFalse(ensemble.contains(addr4.toBookieId())); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - ensembleResponse.isAdheringToPolicy()); + ensembleResponse.getAdheringToPolicy()); } // we could still use addr4 for urgent allocation if it is just bookie @@ -1084,7 +1084,7 @@ public void testPlacementOnStabilizeNetworkTopology() throws Exception { List ensemble = ensembleResponse.getResult(); assertTrue(ensemble.contains(addr4.toBookieId())); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.MEETS_STRICT, - ensembleResponse.isAdheringToPolicy()); + ensembleResponse.getAdheringToPolicy()); } @Test @@ -1135,7 +1135,7 @@ public void testCreateNewEnsembleRandomly() throws Exception { assertFalse("excludeBookie should not be included in the ensemble", newEnsembleSet.contains(addr5.toBookieId())); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.FAIL, - newEnsemblePlacementResult.isAdheringToPolicy()); + newEnsemblePlacementResult.getAdheringToPolicy()); rwAddrs.remove(addr4.toBookieId()); roAddrs.add(addr4.toBookieId()); @@ -1209,7 +1209,7 @@ public void testReplaceBookieRandomly() throws Exception { */ assertEquals("ReplaceBookie candidate", addr7.toBookieId(), replaceBookie); assertEquals("PlacementPolicyAdherence", PlacementPolicyAdherence.FAIL, - replaceResponse.isAdheringToPolicy()); + replaceResponse.getAdheringToPolicy()); rwAddrs.remove(addr7.toBookieId()); excludeBookies.add(addr7.toBookieId()); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java index ff1c71757a6..f8cddf3f5ef 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/AuditorPlacementPolicyCheckTest.java @@ -294,6 +294,151 @@ public void testPlacementPolicyCheckWithLedgersNotAdheringToPlacementPolicy() th } } + @Test + public void testPlacementPolicyCheckWithLedgersNotAdheringToPlacementPolicyAndNotMarkToUnderreplication() + throws Exception { + int numOfBookies = 5; + int numOfLedgersNotAdheringToPlacementPolicy = 0; + List bookieAddresses = new ArrayList<>(); + try (RegistrationManager regManager = driver.createRegistrationManager()) { + for (int i = 0; i < numOfBookies; i++) { + BookieId bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181).toBookieId(); + bookieAddresses.add(bookieAddress); + regManager.registerBookie(bookieAddress, false, BookieServiceInfo.EMPTY); + } + } + + // only three racks + StaticDNSResolver.addNodeToRack("98.98.98.0", "/rack1"); + StaticDNSResolver.addNodeToRack("98.98.98.1", "/rack2"); + StaticDNSResolver.addNodeToRack("98.98.98.2", "/rack3"); + StaticDNSResolver.addNodeToRack("98.98.98.3", "/rack1"); + StaticDNSResolver.addNodeToRack("98.98.98.4", "/rack2"); + + LedgerManagerFactory mFactory = driver.getLedgerManagerFactory(); + LedgerManager lm = mFactory.newLedgerManager(); + int ensembleSize = 5; + int writeQuorumSize = 3; + int ackQuorumSize = 2; + int minNumRacksPerWriteQuorumConfValue = 3; + + /* + * this closed ledger doesn't adhere to placement policy because there are only + * 3 racks, and the ensembleSize is 5. + */ + LedgerMetadata initMeta = LedgerMetadataBuilder.create() + .withId(1L) + .withEnsembleSize(ensembleSize) + .withWriteQuorumSize(writeQuorumSize) + .withAckQuorumSize(ackQuorumSize) + .newEnsembleEntry(0L, bookieAddresses) + .withClosedState() + .withLastEntryId(100) + .withLength(10000) + .withDigestType(DigestType.DUMMY) + .withPassword(new byte[0]) + .build(); + lm.createLedgerMetadata(1L, initMeta).get(); + numOfLedgersNotAdheringToPlacementPolicy++; + + ServerConfiguration servConf = new ServerConfiguration(confByIndex(0)); + servConf.setMinNumRacksPerWriteQuorum(minNumRacksPerWriteQuorumConfValue); + setServerConfigPropertiesForRackPlacement(servConf); + MutableObject auditorRef = new MutableObject(); + try { + TestStatsLogger statsLogger = startAuditorAndWaitForPlacementPolicyCheck(servConf, auditorRef); + Gauge ledgersNotAdheringToPlacementPolicyGuage = statsLogger + .getGauge(ReplicationStats.NUM_LEDGERS_NOT_ADHERING_TO_PLACEMENT_POLICY); + assertEquals("NUM_LEDGERS_NOT_ADHERING_TO_PLACEMENT_POLICY guage value", + numOfLedgersNotAdheringToPlacementPolicy, ledgersNotAdheringToPlacementPolicyGuage.getSample()); + Gauge ledgersSoftlyAdheringToPlacementPolicyGuage = statsLogger + .getGauge(ReplicationStats.NUM_LEDGERS_SOFTLY_ADHERING_TO_PLACEMENT_POLICY); + assertEquals("NUM_LEDGERS_SOFTLY_ADHERING_TO_PLACEMENT_POLICY guage value", + 0, ledgersSoftlyAdheringToPlacementPolicyGuage.getSample()); + } finally { + Auditor auditor = auditorRef.getValue(); + if (auditor != null) { + auditor.close(); + } + } + LedgerUnderreplicationManager underreplicationManager = mFactory.newLedgerUnderreplicationManager(); + long unnderReplicateLedgerId = underreplicationManager.pollLedgerToRereplicate(); + assertEquals(unnderReplicateLedgerId, -1); + } + + @Test + public void testPlacementPolicyCheckWithLedgersNotAdheringToPlacementPolicyAndMarkToUnderreplication() + throws Exception { + int numOfBookies = 5; + int numOfLedgersNotAdheringToPlacementPolicy = 0; + List bookieAddresses = new ArrayList<>(); + try (RegistrationManager regManager = driver.createRegistrationManager()) { + for (int i = 0; i < numOfBookies; i++) { + BookieId bookieAddress = new BookieSocketAddress("98.98.98." + i, 2181).toBookieId(); + bookieAddresses.add(bookieAddress); + regManager.registerBookie(bookieAddress, false, BookieServiceInfo.EMPTY); + } + } + + // only three racks + StaticDNSResolver.addNodeToRack("98.98.98.0", "/rack1"); + StaticDNSResolver.addNodeToRack("98.98.98.1", "/rack2"); + StaticDNSResolver.addNodeToRack("98.98.98.2", "/rack3"); + StaticDNSResolver.addNodeToRack("98.98.98.3", "/rack1"); + StaticDNSResolver.addNodeToRack("98.98.98.4", "/rack2"); + + LedgerManagerFactory mFactory = driver.getLedgerManagerFactory(); + LedgerManager lm = mFactory.newLedgerManager(); + int ensembleSize = 5; + int writeQuorumSize = 3; + int ackQuorumSize = 2; + int minNumRacksPerWriteQuorumConfValue = 3; + + /* + * this closed ledger doesn't adhere to placement policy because there are only + * 3 racks, and the ensembleSize is 5. + */ + LedgerMetadata initMeta = LedgerMetadataBuilder.create() + .withId(1L) + .withEnsembleSize(ensembleSize) + .withWriteQuorumSize(writeQuorumSize) + .withAckQuorumSize(ackQuorumSize) + .newEnsembleEntry(0L, bookieAddresses) + .withClosedState() + .withLastEntryId(100) + .withLength(10000) + .withDigestType(DigestType.DUMMY) + .withPassword(new byte[0]) + .build(); + lm.createLedgerMetadata(1L, initMeta).get(); + numOfLedgersNotAdheringToPlacementPolicy++; + + ServerConfiguration servConf = new ServerConfiguration(confByIndex(0)); + servConf.setMinNumRacksPerWriteQuorum(minNumRacksPerWriteQuorumConfValue); + servConf.setRepairedPlacementPolicyNotAdheringBookieEnable(true); + setServerConfigPropertiesForRackPlacement(servConf); + MutableObject auditorRef = new MutableObject(); + try { + TestStatsLogger statsLogger = startAuditorAndWaitForPlacementPolicyCheck(servConf, auditorRef); + Gauge ledgersNotAdheringToPlacementPolicyGuage = statsLogger + .getGauge(ReplicationStats.NUM_LEDGERS_NOT_ADHERING_TO_PLACEMENT_POLICY); + assertEquals("NUM_LEDGERS_NOT_ADHERING_TO_PLACEMENT_POLICY guage value", + numOfLedgersNotAdheringToPlacementPolicy, ledgersNotAdheringToPlacementPolicyGuage.getSample()); + Gauge ledgersSoftlyAdheringToPlacementPolicyGuage = statsLogger + .getGauge(ReplicationStats.NUM_LEDGERS_SOFTLY_ADHERING_TO_PLACEMENT_POLICY); + assertEquals("NUM_LEDGERS_SOFTLY_ADHERING_TO_PLACEMENT_POLICY guage value", + 0, ledgersSoftlyAdheringToPlacementPolicyGuage.getSample()); + } finally { + Auditor auditor = auditorRef.getValue(); + if (auditor != null) { + auditor.close(); + } + } + LedgerUnderreplicationManager underreplicationManager = mFactory.newLedgerUnderreplicationManager(); + long unnderReplicateLedgerId = underreplicationManager.pollLedgerToRereplicate(); + assertEquals(unnderReplicateLedgerId, 1L); + } + @Test public void testPlacementPolicyCheckForURLedgersElapsedRecoveryGracePeriod() throws Exception { testPlacementPolicyCheckWithURLedgers(true); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java index a69b9cbd491..510d02f8db6 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java @@ -19,32 +19,48 @@ */ package org.apache.bookkeeper.replication; +import static org.apache.bookkeeper.replication.ReplicationStats.AUDITOR_SCOPE; import static org.apache.bookkeeper.replication.ReplicationStats.NUM_ENTRIES_UNABLE_TO_READ_FOR_REPLICATION; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import io.netty.util.HashedWheelTimer; import java.io.IOException; import java.net.URI; +import java.net.UnknownHostException; import java.util.Enumeration; import java.util.List; import java.util.Map.Entry; +import java.util.Objects; import java.util.Optional; import java.util.TimerTask; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; import lombok.Cleanup; +import org.apache.bookkeeper.bookie.BookieImpl; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.client.BookKeeperTestClient; import org.apache.bookkeeper.client.ClientUtil; +import org.apache.bookkeeper.client.EnsemblePlacementPolicy; import org.apache.bookkeeper.client.LedgerEntry; import org.apache.bookkeeper.client.LedgerHandle; +import org.apache.bookkeeper.client.RackawareEnsemblePlacementPolicy; +import org.apache.bookkeeper.client.ZoneawareEnsemblePlacementPolicy; +import org.apache.bookkeeper.client.api.LedgerMetadata; import org.apache.bookkeeper.common.util.OrderedScheduler; +import org.apache.bookkeeper.conf.ClientConfiguration; import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.feature.FeatureProvider; import org.apache.bookkeeper.meta.AbstractZkLedgerManager; import org.apache.bookkeeper.meta.LedgerManager; import org.apache.bookkeeper.meta.LedgerManagerFactory; @@ -53,19 +69,25 @@ import org.apache.bookkeeper.meta.MetadataClientDriver; import org.apache.bookkeeper.meta.MetadataDrivers; import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager; +import org.apache.bookkeeper.meta.exceptions.MetadataException; import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase; import org.apache.bookkeeper.net.BookieId; +import org.apache.bookkeeper.net.DNSToSwitchMapping; +import org.apache.bookkeeper.proto.BookieAddressResolver; import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException; import org.apache.bookkeeper.stats.Counter; +import org.apache.bookkeeper.stats.Gauge; import org.apache.bookkeeper.stats.NullStatsLogger; import org.apache.bookkeeper.stats.StatsLogger; import org.apache.bookkeeper.test.BookKeeperClusterTestCase; import org.apache.bookkeeper.test.TestStatsProvider; import org.apache.bookkeeper.test.TestStatsProvider.TestStatsLogger; import org.apache.bookkeeper.util.BookKeeperConstants; +import org.apache.bookkeeper.util.StaticDNSResolver; import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy; import org.apache.bookkeeper.zookeeper.ZooKeeperClient; import org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase; +import org.apache.commons.lang3.mutable.MutableObject; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.zookeeper.AsyncCallback.StatCallback; import org.apache.zookeeper.KeeperException; @@ -75,6 +97,8 @@ import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.ZooKeeper.States; +import org.apache.zookeeper.data.Stat; +import org.awaitility.Awaitility; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1129,4 +1153,184 @@ public void testRWShutDownInTheCaseOfZKOperationFailures() throws Exception { bkWithMockZK.close(); } } + + @Test + public void testRepairedNotAdheringPlacementPolicyLedgerFragmentsOnRack() throws Exception { + testRepairedNotAdheringPlacementPolicyLedgerFragments(RackawareEnsemblePlacementPolicy.class); + } + + private void testRepairedNotAdheringPlacementPolicyLedgerFragments( + Class placementPolicyClass) throws Exception { + List firstThreeBookies = servers.stream().map(ele -> { + try { + return ele.getServer().getBookieId(); + } catch (UnknownHostException e) { + return null; + } + }).filter(Objects::nonNull).collect(Collectors.toList()); + + baseClientConf.setProperty("reppDnsResolverClass", StaticDNSResolver.class.getName()); + baseClientConf.setProperty("enforceStrictZoneawarePlacement", false); + bkc = new BookKeeperTestClient(baseClientConf) { + @Override + protected EnsemblePlacementPolicy initializeEnsemblePlacementPolicy(ClientConfiguration conf, + DNSToSwitchMapping dnsResolver, HashedWheelTimer timer, FeatureProvider featureProvider, + StatsLogger statsLogger, BookieAddressResolver bookieAddressResolver) throws IOException { + EnsemblePlacementPolicy ensemblePlacementPolicy = null; + if (ZoneawareEnsemblePlacementPolicy.class == placementPolicyClass) { + ensemblePlacementPolicy = buildZoneAwareEnsemblePlacementPolicy(firstThreeBookies); + } else if (RackawareEnsemblePlacementPolicy.class == placementPolicyClass) { + ensemblePlacementPolicy = buildRackAwareEnsemblePlacementPolicy(firstThreeBookies); + } + ensemblePlacementPolicy.initialize(conf, Optional.ofNullable(dnsResolver), timer, + featureProvider, statsLogger, bookieAddressResolver); + return ensemblePlacementPolicy; + } + }; + + //This ledger not adhering placement policy, the combine(0,1,2) rack is 1. + LedgerHandle lh = bkc.createLedger(3, 3, 3, BookKeeper.DigestType.CRC32, TESTPASSWD); + + int entrySize = 10; + for (int i = 0; i < entrySize; i++) { + lh.addEntry(data); + } + lh.close(); + + int minNumRacksPerWriteQuorumConfValue = 2; + + ServerConfiguration servConf = new ServerConfiguration(confByIndex(0)); + servConf.setMinNumRacksPerWriteQuorum(minNumRacksPerWriteQuorumConfValue); + servConf.setProperty("reppDnsResolverClass", StaticDNSResolver.class.getName()); + servConf.setAuditorPeriodicPlacementPolicyCheckInterval(1000); + servConf.setRepairedPlacementPolicyNotAdheringBookieEnable(true); + + MutableObject auditorRef = new MutableObject(); + try { + TestStatsLogger statsLogger = startAuditorAndWaitForPlacementPolicyCheck(servConf, auditorRef); + Gauge ledgersNotAdheringToPlacementPolicyGuage = statsLogger + .getGauge(ReplicationStats.NUM_LEDGERS_NOT_ADHERING_TO_PLACEMENT_POLICY); + assertEquals("NUM_LEDGERS_NOT_ADHERING_TO_PLACEMENT_POLICY guage value", + 1, ledgersNotAdheringToPlacementPolicyGuage.getSample()); + Gauge ledgersSoftlyAdheringToPlacementPolicyGuage = statsLogger + .getGauge(ReplicationStats.NUM_LEDGERS_SOFTLY_ADHERING_TO_PLACEMENT_POLICY); + assertEquals("NUM_LEDGERS_SOFTLY_ADHERING_TO_PLACEMENT_POLICY guage value", + 0, ledgersSoftlyAdheringToPlacementPolicyGuage.getSample()); + } finally { + Auditor auditor = auditorRef.getValue(); + if (auditor != null) { + auditor.close(); + } + } + + Stat stat = bkc.getZkHandle() + .exists("/ledgers/underreplication/ledgers/0000/0000/0000/0000/urL0000000000", false); + assertNotNull(stat); + + baseConf.setRepairedPlacementPolicyNotAdheringBookieEnable(true); + BookKeeper bookKeeper = new BookKeeperTestClient(baseClientConf) { + @Override + protected EnsemblePlacementPolicy initializeEnsemblePlacementPolicy(ClientConfiguration conf, + DNSToSwitchMapping dnsResolver, HashedWheelTimer timer, FeatureProvider featureProvider, + StatsLogger statsLogger, BookieAddressResolver bookieAddressResolver) throws IOException { + EnsemblePlacementPolicy ensemblePlacementPolicy = null; + if (ZoneawareEnsemblePlacementPolicy.class == placementPolicyClass) { + ensemblePlacementPolicy = buildZoneAwareEnsemblePlacementPolicy(firstThreeBookies); + } else if (RackawareEnsemblePlacementPolicy.class == placementPolicyClass) { + ensemblePlacementPolicy = buildRackAwareEnsemblePlacementPolicy(firstThreeBookies); + } + ensemblePlacementPolicy.initialize(conf, Optional.ofNullable(dnsResolver), timer, + featureProvider, statsLogger, bookieAddressResolver); + return ensemblePlacementPolicy; + } + }; + ReplicationWorker rw = new ReplicationWorker(baseConf, bookKeeper, false, NullStatsLogger.INSTANCE); + rw.start(); + + //start new bookie, the rack is /rack2 + BookieId newBookieId = startNewBookieAndReturnBookieId(); + + Awaitility.await().untilAsserted(() -> { + LedgerMetadata metadata = bkc.getLedgerManager().readLedgerMetadata(lh.getId()).get().getValue(); + List newBookies = metadata.getAllEnsembles().get(0L); + assertTrue(newBookies.contains(newBookieId)); + }); + + Awaitility.await().untilAsserted(() -> { + Stat stat1 = bkc.getZkHandle() + .exists("/ledgers/underreplication/ledgers/0000/0000/0000/0000/urL0000000000", false); + assertNull(stat1); + }); + + for (BookieId rack1Book : firstThreeBookies) { + killBookie(rack1Book); + } + + verifyRecoveredLedgers(lh, 0, entrySize - 1); + } + + private EnsemblePlacementPolicy buildRackAwareEnsemblePlacementPolicy(List bookieIds) { + return new RackawareEnsemblePlacementPolicy() { + @Override + public String resolveNetworkLocation(BookieId addr) { + if (bookieIds.contains(addr)) { + return "/rack1"; + } + //The other bookie is /rack2 + return "/rack2"; + } + }; + } + + private EnsemblePlacementPolicy buildZoneAwareEnsemblePlacementPolicy(List firstThreeBookies) { + return new ZoneawareEnsemblePlacementPolicy() { + @Override + protected String resolveNetworkLocation(BookieId addr) { + //The first three bookie 1 is /zone1/ud1 + //The first three bookie 2,3 is /zone1/ud2 + if (firstThreeBookies.get(0).equals(addr)) { + return "/zone1/ud1"; + } else if (firstThreeBookies.contains(addr)) { + return "/zone1/ud2"; + } + //The other bookie is /zone2/ud1 + return "/zone2/ud1"; + } + }; + } + + private TestStatsLogger startAuditorAndWaitForPlacementPolicyCheck(ServerConfiguration servConf, + MutableObject auditorRef) throws MetadataException, CompatibilityException, KeeperException, + InterruptedException, ReplicationException.UnavailableException, UnknownHostException { + LedgerManagerFactory mFactory = driver.getLedgerManagerFactory(); + LedgerUnderreplicationManager urm = mFactory.newLedgerUnderreplicationManager(); + TestStatsProvider statsProvider = new TestStatsProvider(); + TestStatsLogger statsLogger = statsProvider.getStatsLogger(AUDITOR_SCOPE); + TestStatsProvider.TestOpStatsLogger placementPolicyCheckStatsLogger = + (TestStatsProvider.TestOpStatsLogger) statsLogger + .getOpStatsLogger(ReplicationStats.PLACEMENT_POLICY_CHECK_TIME); + + final AuditorPeriodicCheckTest.TestAuditor auditor = new AuditorPeriodicCheckTest.TestAuditor( + BookieImpl.getBookieId(servConf).toString(), servConf, bkc, false, statsLogger); + auditorRef.setValue(auditor); + CountDownLatch latch = auditor.getLatch(); + assertEquals("PLACEMENT_POLICY_CHECK_TIME SuccessCount", 0, + placementPolicyCheckStatsLogger.getSuccessCount()); + urm.setPlacementPolicyCheckCTime(-1); + auditor.start(); + /* + * since placementPolicyCheckCTime is set to -1, placementPolicyCheck should be + * scheduled to run with no initialdelay + */ + assertTrue("placementPolicyCheck should have executed", latch.await(20, TimeUnit.SECONDS)); + for (int i = 0; i < 20; i++) { + Thread.sleep(100); + if (placementPolicyCheckStatsLogger.getSuccessCount() >= 1) { + break; + } + } + assertEquals("PLACEMENT_POLICY_CHECK_TIME SuccessCount", 1, + placementPolicyCheckStatsLogger.getSuccessCount()); + return statsLogger; + } } diff --git a/site3/website/docs/reference/config.md b/site3/website/docs/reference/config.md index 57d9abd89bc..232100a9701 100644 --- a/site3/website/docs/reference/config.md +++ b/site3/website/docs/reference/config.md @@ -325,16 +325,17 @@ The table below lists parameters that you can set to configure bookies. All conf ## AutoRecovery auditor settings -| Parameter | Description | Default -| --------- | ----------- | ------- | -| auditorPeriodicBookieCheckInterval | The time interval between auditor bookie checks, in seconds. The auditor bookie check checks ledger metadata to see which bookies should contain entries for each ledger. If a bookie that should contain entries is unavailable, then the ledger containing that entry is marked for recovery. Setting this to 0 disables the periodic check. Bookie checks will still run when a bookie fails. The default is once per day. | 86400 | -| auditorPeriodicCheckInterval | The time interval, in seconds, at which the auditor will check all ledgers in the cluster. By default this runs once a week.

Set this to 0 to disable the periodic check completely. Note that periodic checking will put extra load on the cluster, so it should not be run more frequently than once a day.
| 604800 | -| auditorPeriodicPlacementPolicyCheckInterval | The time interval between auditor placement policy checks, in seconds. The auditor placement policy check validates if the ensemble of segments of all the closed ledgers is adhering to the placement policy. It is just monitoring scrutiny but doesn't take any corrective measure other than logging error and reporting metrics. By default it is disabled. | | -| auditorLedgerVerificationPercentage | The percentage of a ledger (fragment)'s entries will be verified before claiming a fragment as missing. If it is 0, it only verifies the first and last entries of a given fragment.
| | -| lostBookieRecoveryDelay | How long to wait, in seconds, before starting autorecovery of a lost bookie. | | -| storeSystemTimeAsLedgerUnderreplicatedMarkTime | Enable the Auditor to use system time as underreplicated ledger mark time. If this is enabled, Auditor will write a ctime field into the underreplicated ledger znode. | true | -| underreplicatedLedgerRecoveryGracePeriod | The grace period (in seconds) for underreplicated ledgers recovery. If ledger is marked underreplicated for more than this period then it will be reported by placementPolicyCheck in Auditor. Setting this to 0 will disable this check. | | -| auditorReplicasCheckInterval | Sets the regularity/interval at which the auditor will run a replicas check of all ledgers, which are closed. This should not be run very often since it validates availability of replicas of all ledgers by querying bookies. Setting this to 0 will completely disable the periodic replicas check. By default it is disabled. | | +| Parameter | Description | Default +| --------- |----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------| +| auditorPeriodicBookieCheckInterval | The time interval between auditor bookie checks, in seconds. The auditor bookie check checks ledger metadata to see which bookies should contain entries for each ledger. If a bookie that should contain entries is unavailable, then the ledger containing that entry is marked for recovery. Setting this to 0 disables the periodic check. Bookie checks will still run when a bookie fails. The default is once per day. | 86400 | +| auditorPeriodicCheckInterval | The time interval, in seconds, at which the auditor will check all ledgers in the cluster. By default this runs once a week.

Set this to 0 to disable the periodic check completely. Note that periodic checking will put extra load on the cluster, so it should not be run more frequently than once a day.
| 604800 | +| auditorPeriodicPlacementPolicyCheckInterval | The time interval between auditor placement policy checks, in seconds. The auditor placement policy check validates if the ensemble of segments of all the closed ledgers is adhering to the placement policy. It is just monitoring scrutiny but doesn't take any corrective measure other than logging error and reporting metrics. By default, it is disabled. | 0 | +| repairedPlacementPolicyNotAdheringBookieEnabled | In Auditor, it combines with auditorPeriodicPlacementPolicyCheckInterval, to control is marked ledger id to under replication managed when found a ledger ensemble not adhere to placement policy. In ReplicationWorker, to control is to repair the ledger which the ensemble does not adhere to the placement policy. By default, it is disabled. If you want to enable this feature, consider two factors. 1:Must config RackawareEnsemblePlacementPolicy. 2:There maybe lots of ledger will be mark underreplicated. The replicationWorker will replicate lots of ledger, it will increase read request and write request in bookie server. You should set a suitable rereplicationEntryBatchSize to avoid bookie server pressure. | false | +| auditorLedgerVerificationPercentage | The percentage of a ledger (fragment)'s entries will be verified before claiming a fragment as missing. If it is 0, it only verifies the first and last entries of a given fragment.
| | +| lostBookieRecoveryDelay | How long to wait, in seconds, before starting autorecovery of a lost bookie. | | +| storeSystemTimeAsLedgerUnderreplicatedMarkTime | Enable the Auditor to use system time as underreplicated ledger mark time. If this is enabled, Auditor will write a ctime field into the underreplicated ledger znode. | true | +| underreplicatedLedgerRecoveryGracePeriod | The grace period (in seconds) for underreplicated ledgers recovery. If ledger is marked underreplicated for more than this period then it will be reported by placementPolicyCheck in Auditor. Setting this to 0 will disable this check. | | +| auditorReplicasCheckInterval | Sets the regularity/interval at which the auditor will run a replicas check of all ledgers, which are closed. This should not be run very often since it validates availability of replicas of all ledgers by querying bookies. Setting this to 0 will completely disable the periodic replicas check. By default it is disabled. | | ## AutoRecovery replication worker settings