|
46 | 46 | import java.util.concurrent.TimeUnit; |
47 | 47 |
|
48 | 48 | import java.util.concurrent.locks.ReentrantReadWriteLock; |
| 49 | +import java.util.stream.Collectors; |
| 50 | +import java.util.stream.Stream; |
49 | 51 | import org.apache.bookkeeper.client.BKException.BKNotEnoughBookiesException; |
50 | 52 | import org.apache.bookkeeper.client.BookieInfoReader.BookieInfo; |
51 | 53 | import org.apache.bookkeeper.client.WeightedRandomSelection.WeightedObject; |
|
69 | 71 | import org.apache.bookkeeper.stats.StatsLogger; |
70 | 72 | import org.apache.bookkeeper.stats.annotations.StatsDoc; |
71 | 73 | import org.apache.commons.collections4.CollectionUtils; |
| 74 | +import org.apache.commons.lang3.tuple.Pair; |
72 | 75 | import org.slf4j.Logger; |
73 | 76 | import org.slf4j.LoggerFactory; |
74 | 77 |
|
@@ -1071,4 +1074,175 @@ public boolean areAckedBookiesAdheringToPlacementPolicy(Set<BookieId> ackedBooki |
1071 | 1074 | } |
1072 | 1075 | return rackCounter.size() >= minWriteQuorumNumRacksPerWriteQuorum; |
1073 | 1076 | } |
| 1077 | + |
| 1078 | + @Override |
| 1079 | + public PlacementResult<List<BookieId>> replaceToAdherePlacementPolicy( |
| 1080 | + int ensembleSize, |
| 1081 | + int writeQuorumSize, |
| 1082 | + int ackQuorumSize, |
| 1083 | + Set<BookieId> excludeBookies, |
| 1084 | + List<BookieId> currentEnsemble) { |
| 1085 | + rwLock.readLock().lock(); |
| 1086 | + try { |
| 1087 | + final List<BookieNode> provisionalEnsembleNodes = currentEnsemble.stream() |
| 1088 | + .map(this::convertBookieToNode).collect(Collectors.toList()); |
| 1089 | + final Set<Node> excludeNodes = convertBookiesToNodes( |
| 1090 | + addDefaultRackBookiesIfMinNumRacksIsEnforced(excludeBookies)); |
| 1091 | + int minNumRacksPerWriteQuorumForThisEnsemble = Math.min(writeQuorumSize, minNumRacksPerWriteQuorum); |
| 1092 | + final RRTopologyAwareCoverageEnsemble ensemble = |
| 1093 | + new RRTopologyAwareCoverageEnsemble( |
| 1094 | + ensembleSize, |
| 1095 | + writeQuorumSize, |
| 1096 | + ackQuorumSize, |
| 1097 | + RACKNAME_DISTANCE_FROM_LEAVES, |
| 1098 | + null, |
| 1099 | + null, |
| 1100 | + minNumRacksPerWriteQuorumForThisEnsemble); |
| 1101 | + |
| 1102 | + int numRacks = topology.getNumOfRacks(); |
| 1103 | + // only one rack or less than minNumRacksPerWriteQuorumForThisEnsemble, stop calculation to skip relocation |
| 1104 | + if (numRacks < 2 || numRacks < minNumRacksPerWriteQuorumForThisEnsemble) { |
| 1105 | + LOG.warn("Skip ensemble relocation because the cluster has only {} rack.", numRacks); |
| 1106 | + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); |
| 1107 | + } |
| 1108 | + |
| 1109 | + BookieNode prevNode = null; |
| 1110 | + final BookieNode firstNode = provisionalEnsembleNodes.get(0); |
| 1111 | + // use same bookie at first to reduce ledger replication |
| 1112 | + if (!excludeNodes.contains(firstNode) && ensemble.apply(firstNode, ensemble) |
| 1113 | + && ensemble.addNode(firstNode)) { |
| 1114 | + excludeNodes.add(firstNode); |
| 1115 | + prevNode = firstNode; |
| 1116 | + } |
| 1117 | + |
| 1118 | + for (int i = prevNode == null ? 0 : 1; i < ensembleSize; i++) { |
| 1119 | + final String curRack; |
| 1120 | + if (null == prevNode) { |
| 1121 | + if ((null == localNode) || defaultRack.equals(localNode.getNetworkLocation())) { |
| 1122 | + curRack = NodeBase.ROOT; |
| 1123 | + } else { |
| 1124 | + curRack = localNode.getNetworkLocation(); |
| 1125 | + } |
| 1126 | + } else { |
| 1127 | + curRack = "~" + prevNode.getNetworkLocation(); |
| 1128 | + } |
| 1129 | + |
| 1130 | + try { |
| 1131 | + prevNode = replaceToAdherePlacementPolicyInternal( |
| 1132 | + curRack, excludeNodes, ensemble, ensemble, |
| 1133 | + provisionalEnsembleNodes, i, ensembleSize, minNumRacksPerWriteQuorumForThisEnsemble); |
| 1134 | + // replace to newer node |
| 1135 | + provisionalEnsembleNodes.set(i, prevNode); |
| 1136 | + } catch (BKNotEnoughBookiesException e) { |
| 1137 | + LOG.warn("Skip ensemble relocation because the cluster has not enough bookies."); |
| 1138 | + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); |
| 1139 | + } |
| 1140 | + } |
| 1141 | + List<BookieId> bookieList = ensemble.toList(); |
| 1142 | + if (ensembleSize != bookieList.size()) { |
| 1143 | + LOG.warn("Not enough {} bookies are available to form an ensemble : {}.", |
| 1144 | + ensembleSize, bookieList); |
| 1145 | + return PlacementResult.of(Collections.emptyList(), PlacementPolicyAdherence.FAIL); |
| 1146 | + } |
| 1147 | + return PlacementResult.of(bookieList, |
| 1148 | + isEnsembleAdheringToPlacementPolicy( |
| 1149 | + bookieList, writeQuorumSize, ackQuorumSize)); |
| 1150 | + } finally { |
| 1151 | + rwLock.readLock().unlock(); |
| 1152 | + } |
| 1153 | + } |
| 1154 | + |
| 1155 | + private BookieNode replaceToAdherePlacementPolicyInternal( |
| 1156 | + String netPath, Set<Node> excludeBookies, Predicate<BookieNode> predicate, |
| 1157 | + Ensemble<BookieNode> ensemble, List<BookieNode> provisionalEnsembleNodes, int ensembleIndex, |
| 1158 | + int ensembleSize, int minNumRacksPerWriteQuorumForThisEnsemble) throws BKNotEnoughBookiesException { |
| 1159 | + final BookieNode currentNode = provisionalEnsembleNodes.get(ensembleIndex); |
| 1160 | + // if the current bookie could be applied to the ensemble, apply it to minify the number of bookies replaced |
| 1161 | + if (!excludeBookies.contains(currentNode) && predicate.apply(currentNode, ensemble)) { |
| 1162 | + if (ensemble.addNode(currentNode)) { |
| 1163 | + // add the candidate to exclude set |
| 1164 | + excludeBookies.add(currentNode); |
| 1165 | + } |
| 1166 | + return currentNode; |
| 1167 | + } |
| 1168 | + |
| 1169 | + final List<Pair<String, List<BookieNode>>> conditionList = new ArrayList<>(); |
| 1170 | + final Set<String> preExcludeRacks = new HashSet<>(); |
| 1171 | + final Set<String> postExcludeRacks = new HashSet<>(); |
| 1172 | + for (int i = 0; i < minNumRacksPerWriteQuorumForThisEnsemble - 1; i++) { |
| 1173 | + preExcludeRacks.add(provisionalEnsembleNodes.get(Math.floorMod((ensembleIndex - i - 1), ensembleSize)) |
| 1174 | + .getNetworkLocation()); |
| 1175 | + postExcludeRacks.add(provisionalEnsembleNodes.get(Math.floorMod((ensembleIndex + i + 1), ensembleSize)) |
| 1176 | + .getNetworkLocation()); |
| 1177 | + } |
| 1178 | + // adhere minNumRacksPerWriteQuorum by preExcludeRacks |
| 1179 | + // avoid additional replace from write quorum candidates by preExcludeRacks and postExcludeRacks |
| 1180 | + // avoid to use first candidate bookies for election by provisionalEnsembleNodes |
| 1181 | + conditionList.add(Pair.of( |
| 1182 | + "~" + String.join(",", |
| 1183 | + Stream.concat(preExcludeRacks.stream(), postExcludeRacks.stream()).collect(Collectors.toSet())), |
| 1184 | + provisionalEnsembleNodes |
| 1185 | + )); |
| 1186 | + // avoid to use same rack between previous index by netPath |
| 1187 | + // avoid to use first candidate bookies for election by provisionalEnsembleNodes |
| 1188 | + conditionList.add(Pair.of(netPath, provisionalEnsembleNodes)); |
| 1189 | + // avoid to use same rack between previous index by netPath |
| 1190 | + conditionList.add(Pair.of(netPath, Collections.emptyList())); |
| 1191 | + |
| 1192 | + for (Pair<String, List<BookieNode>> condition : conditionList) { |
| 1193 | + WeightedRandomSelection<BookieNode> wRSelection = null; |
| 1194 | + |
| 1195 | + final List<Node> leaves = new ArrayList<>(topology.getLeaves(condition.getLeft())); |
| 1196 | + if (!isWeighted) { |
| 1197 | + Collections.shuffle(leaves); |
| 1198 | + } else { |
| 1199 | + if (CollectionUtils.subtract(leaves, excludeBookies).size() < 1) { |
| 1200 | + throw new BKNotEnoughBookiesException(); |
| 1201 | + } |
| 1202 | + wRSelection = prepareForWeightedSelection(leaves); |
| 1203 | + if (wRSelection == null) { |
| 1204 | + throw new BKNotEnoughBookiesException(); |
| 1205 | + } |
| 1206 | + } |
| 1207 | + |
| 1208 | + final Iterator<Node> it = leaves.iterator(); |
| 1209 | + final Set<Node> bookiesSeenSoFar = new HashSet<>(); |
| 1210 | + while (true) { |
| 1211 | + Node n; |
| 1212 | + if (isWeighted) { |
| 1213 | + if (bookiesSeenSoFar.size() == leaves.size()) { |
| 1214 | + // Don't loop infinitely. |
| 1215 | + break; |
| 1216 | + } |
| 1217 | + n = wRSelection.getNextRandom(); |
| 1218 | + bookiesSeenSoFar.add(n); |
| 1219 | + } else { |
| 1220 | + if (it.hasNext()) { |
| 1221 | + n = it.next(); |
| 1222 | + } else { |
| 1223 | + break; |
| 1224 | + } |
| 1225 | + } |
| 1226 | + if (excludeBookies.contains(n)) { |
| 1227 | + continue; |
| 1228 | + } |
| 1229 | + if (!(n instanceof BookieNode) || !predicate.apply((BookieNode) n, ensemble)) { |
| 1230 | + continue; |
| 1231 | + } |
| 1232 | + // additional excludeBookies |
| 1233 | + if (condition.getRight().contains(n)) { |
| 1234 | + continue; |
| 1235 | + } |
| 1236 | + BookieNode bn = (BookieNode) n; |
| 1237 | + // got a good candidate |
| 1238 | + if (ensemble.addNode(bn)) { |
| 1239 | + // add the candidate to exclude set |
| 1240 | + excludeBookies.add(bn); |
| 1241 | + } |
| 1242 | + return bn; |
| 1243 | + } |
| 1244 | + } |
| 1245 | + |
| 1246 | + throw new BKNotEnoughBookiesException(); |
| 1247 | + } |
1074 | 1248 | } |
0 commit comments