Skip to content

Commit d49826f

Browse files
equanzequanz
committed
fix: move relocation procecure to BookKeeperAdmin
1 parent e6e4deb commit d49826f

File tree

4 files changed

+257
-260
lines changed

4 files changed

+257
-260
lines changed

‎bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperAdmin.java‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import static com.google.common.base.Preconditions.checkArgument;
2424
import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithMetadataBookieDriver;
2525
import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithRegistrationManager;
26+
import com.google.common.base.Functions;
2627
import com.google.common.collect.Lists;
2728
import com.google.common.collect.Maps;
2829
import com.google.common.collect.Sets;
@@ -31,6 +32,7 @@
3132
import java.io.IOException;
3233
import java.util.ArrayList;
3334
import java.util.Collection;
35+
import java.util.Collections;
3436
import java.util.Enumeration;
3537
import java.util.HashMap;
3638
import java.util.Iterator;
@@ -50,6 +52,8 @@
5052
import java.util.concurrent.atomic.AtomicReference;
5153
import java.util.function.BiConsumer;
5254
import java.util.function.Predicate;
55+
import java.util.stream.Collectors;
56+
import java.util.stream.IntStream;
5357
import lombok.SneakyThrows;
5458
import org.apache.bookkeeper.bookie.BookieException;
5559
import org.apache.bookkeeper.bookie.BookieImpl;
@@ -1225,6 +1229,108 @@ public void processResult(int rc, String s, Object ctx) {
12251229
}
12261230
}
12271231

1232+
/**
1233+
*
1234+
* @param lh Ledger Handle
1235+
* @param dryRun if true, run it without any modification.
1236+
* @return failed ledger fragment indices
1237+
* @throws UnsupportedOperationException Default behavior of
1238+
* {@link EnsemblePlacementPolicy#replaceToAdherePlacementPolicy(int, int, int, java.util.Set, java.util.List)}.
1239+
*/
1240+
public List<Long> relocateLedgerToAdherePlacementPolicy(LedgerHandle lh, boolean dryRun)
1241+
throws UnsupportedOperationException {
1242+
final EnsemblePlacementPolicy placementPolicy = bkc.getPlacementPolicy();
1243+
1244+
final long ledgerId = lh.getId();
1245+
final LedgerMetadata ledgerMeta = lh.getLedgerMetadata();
1246+
final List<Long> failedFragmentIndexList = new ArrayList<>();
1247+
final Map<Long, Long> ledgerFragmentsRange = new HashMap<>();
1248+
Long curEntryId = null;
1249+
for (Map.Entry<Long, ? extends List<BookieId>> entry :
1250+
ledgerMeta.getAllEnsembles().entrySet()) {
1251+
if (curEntryId != null) {
1252+
ledgerFragmentsRange.put(curEntryId, entry.getKey() - 1);
1253+
}
1254+
curEntryId = entry.getKey();
1255+
}
1256+
if (curEntryId != null) {
1257+
ledgerFragmentsRange.put(curEntryId, lh.getLastAddConfirmed());
1258+
}
1259+
1260+
for (Map.Entry<Long, ? extends List<BookieId>> entry : ledgerMeta.getAllEnsembles().entrySet()) {
1261+
if (placementPolicy.isEnsembleAdheringToPlacementPolicy(entry.getValue(),
1262+
ledgerMeta.getWriteQuorumSize(), ledgerMeta.getAckQuorumSize())
1263+
== EnsemblePlacementPolicy.PlacementPolicyAdherence.FAIL) {
1264+
final List<BookieId> currentEnsemble = entry.getValue();
1265+
// Currently, don't consider quarantinedBookies
1266+
final EnsemblePlacementPolicy.PlacementResult<List<BookieId>> placementResult =
1267+
placementPolicy.replaceToAdherePlacementPolicy(
1268+
ledgerMeta.getEnsembleSize(),
1269+
ledgerMeta.getWriteQuorumSize(),
1270+
ledgerMeta.getAckQuorumSize(),
1271+
Collections.emptySet(),
1272+
currentEnsemble);
1273+
1274+
if (placementResult.isAdheringToPolicy()
1275+
== EnsemblePlacementPolicy.PlacementPolicyAdherence.FAIL) {
1276+
LOG.warn("Failed to relocate the ensemble. So, skip the operation."
1277+
+ " ledgerId: {}, fragmentIndex: {}",
1278+
ledgerId, entry.getKey());
1279+
failedFragmentIndexList.add(entry.getKey());
1280+
} else {
1281+
final List<BookieId> newEnsemble = placementResult.getResult();
1282+
final Map<Integer, BookieId> replaceBookiesMap = IntStream
1283+
.range(0, ledgerMeta.getEnsembleSize()).boxed()
1284+
.filter(i -> !newEnsemble.get(i).equals(currentEnsemble.get(i)))
1285+
.collect(Collectors.toMap(Functions.identity(), newEnsemble::get));
1286+
if (replaceBookiesMap.isEmpty()) {
1287+
LOG.warn("Failed to get bookies to replace. So, skip the operation."
1288+
+ " ledgerId: {}, fragmentIndex: {}",
1289+
ledgerId, entry.getKey());
1290+
failedFragmentIndexList.add(entry.getKey());
1291+
} else if (dryRun) {
1292+
LOG.info("Would replace the ensemble. ledgerId: {}, fragmentIndex: {},"
1293+
+ " currentEnsemble: {} replaceBookiesMap {}",
1294+
ledgerId, entry.getKey(),
1295+
currentEnsemble, replaceBookiesMap);
1296+
} else {
1297+
if (LOG.isDebugEnabled()) {
1298+
LOG.debug("Try to replace the ensemble. ledgerId: {}, fragmentIndex: {},"
1299+
+ " replaceBookiesMap {}",
1300+
ledgerId, entry.getKey(), replaceBookiesMap);
1301+
}
1302+
final LedgerFragment fragment = new LedgerFragment(lh, entry.getKey(),
1303+
ledgerFragmentsRange.get(entry.getKey()), replaceBookiesMap.keySet());
1304+
1305+
try {
1306+
replicateLedgerFragment(lh, fragment, replaceBookiesMap,
1307+
(lId, eId) -> {
1308+
// This consumer is already accepted before the method returns
1309+
// void. Therefore, use failedFragmentIndexList in this consumer.
1310+
LOG.warn("Failed to read entry {}:{}", lId, eId);
1311+
failedFragmentIndexList.add(entry.getKey());
1312+
});
1313+
if (LOG.isDebugEnabled()) {
1314+
LOG.debug("Operation finished in the ensemble. ledgerId: {},"
1315+
+ " fragmentIndex: {}, replaceBookiesMap {}",
1316+
ledgerId, entry.getKey(), replaceBookiesMap);
1317+
}
1318+
} catch (BKException | InterruptedException e) {
1319+
LOG.warn("Failed to replicate ledger fragment.", e);
1320+
failedFragmentIndexList.add(entry.getKey());
1321+
}
1322+
}
1323+
}
1324+
} else {
1325+
if (LOG.isDebugEnabled()) {
1326+
LOG.debug("The fragment is adhering to placement policy. So, skip the operation."
1327+
+ " ledgerId: {}, fragmentIndex: {}", ledgerId, entry.getKey());
1328+
}
1329+
}
1330+
}
1331+
return failedFragmentIndexList;
1332+
}
1333+
12281334
/**
12291335
* Format the BookKeeper metadata in zookeeper.
12301336
*

‎bookkeeper-server/src/main/java/org/apache/bookkeeper/tools/cli/commands/bookies/CorrectEnsemblePlacementCommand.java‎

Lines changed: 11 additions & 104 deletions
Original file line numberDiff line numberDiff line change
@@ -20,38 +20,29 @@
2020
import com.beust.jcommander.Parameter;
2121
import com.beust.jcommander.converters.CommaParameterSplitter;
2222
import com.google.common.annotations.VisibleForTesting;
23-
import com.google.common.base.Functions;
2423
import com.google.common.util.concurrent.UncheckedExecutionException;
2524
import java.io.IOException;
2625
import java.net.URI;
27-
import java.util.Collections;
28-
import java.util.HashMap;
2926
import java.util.List;
30-
import java.util.Map;
3127
import java.util.NavigableSet;
3228
import java.util.TreeSet;
3329
import java.util.concurrent.ConcurrentSkipListSet;
3430
import java.util.concurrent.CountDownLatch;
3531
import java.util.stream.Collectors;
36-
import java.util.stream.IntStream;
3732
import lombok.Cleanup;
3833
import lombok.Setter;
3934
import lombok.experimental.Accessors;
4035
import org.apache.bookkeeper.bookie.BookieException;
4136
import org.apache.bookkeeper.client.BKException;
4237
import org.apache.bookkeeper.client.BookKeeper;
4338
import org.apache.bookkeeper.client.BookKeeperAdmin;
44-
import org.apache.bookkeeper.client.EnsemblePlacementPolicy;
45-
import org.apache.bookkeeper.client.LedgerFragment;
46-
import org.apache.bookkeeper.client.api.LedgerMetadata;
4739
import org.apache.bookkeeper.conf.ClientConfiguration;
4840
import org.apache.bookkeeper.conf.ServerConfiguration;
4941
import org.apache.bookkeeper.meta.LedgerManagerFactory;
5042
import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
5143
import org.apache.bookkeeper.meta.MetadataBookieDriver;
5244
import org.apache.bookkeeper.meta.MetadataDrivers;
5345
import org.apache.bookkeeper.meta.exceptions.MetadataException;
54-
import org.apache.bookkeeper.net.BookieId;
5546
import org.apache.bookkeeper.replication.ReplicationException;
5647
import org.apache.bookkeeper.stats.NullStatsLogger;
5748
import org.apache.bookkeeper.tools.cli.helpers.BookieCommand;
@@ -133,7 +124,7 @@ public boolean apply(ServerConfiguration conf, CorrectEnsemblePlacementFlags fla
133124

134125
final ClientConfiguration clientConf = new ClientConfiguration(conf);
135126
final BookKeeper bookKeeper = new BookKeeper(clientConf);
136-
final BookKeeperAdmin admin = new BookKeeperAdmin(clientConf);
127+
final BookKeeperAdmin admin = new BookKeeperAdmin(bookKeeper);
137128
return relocate(conf, flags, bookKeeper, admin);
138129
} catch (Exception e) {
139130
throw new UncheckedExecutionException(e.getMessage(), e);
@@ -143,8 +134,6 @@ public boolean apply(ServerConfiguration conf, CorrectEnsemblePlacementFlags fla
143134
@VisibleForTesting
144135
public boolean relocate(ServerConfiguration conf, CorrectEnsemblePlacementFlags flags,
145136
BookKeeper bookKeeper, BookKeeperAdmin admin) throws Exception {
146-
final EnsemblePlacementPolicy placementPolicy = bookKeeper.getPlacementPolicy();
147-
148137
@Cleanup
149138
final MetadataBookieDriver metadataDriver = instantiateMetadataDriver(conf);
150139
@Cleanup
@@ -153,7 +142,7 @@ public boolean relocate(ServerConfiguration conf, CorrectEnsemblePlacementFlags
153142
final LedgerUnderreplicationManager lum = lmf.newLedgerUnderreplicationManager();
154143

155144
final List<Long> targetLedgers =
156-
flags.ledgerIds.stream().parallel().distinct().filter(ledgerId -> {
145+
flags.ledgerIds.stream().distinct().filter(ledgerId -> {
157146
try {
158147
return (!flags.skipOpenLedgers || bookKeeper.isClosed(ledgerId))
159148
&& !lum.isLedgerBeingReplicated(ledgerId);
@@ -172,6 +161,9 @@ public boolean relocate(ServerConfiguration conf, CorrectEnsemblePlacementFlags
172161
final NavigableSet<Pair<Long, Long>> failedTargets = new ConcurrentSkipListSet<>();
173162
final CountDownLatch latch = new CountDownLatch(targetLedgers.size());
174163
for (long ledgerId : targetLedgers) {
164+
if (LOG.isDebugEnabled()) {
165+
LOG.debug("Start relocation of the ledger {}.", ledgerId);
166+
}
175167
if (!flags.dryRun) {
176168
try {
177169
lum.acquireUnderreplicatedLedger(ledgerId);
@@ -188,103 +180,18 @@ public boolean relocate(ServerConfiguration conf, CorrectEnsemblePlacementFlags
188180
LOG.warn("Failed to open ledger {}", ledgerId);
189181
return;
190182
}
191-
192-
final LedgerMetadata ledgerMeta = lh.getLedgerMetadata();
193-
final Map<Long, Long> ledgerFragmentsRange = new HashMap<>();
194-
Long curEntryId = null;
195-
for (Map.Entry<Long, ? extends List<BookieId>> entry :
196-
ledgerMeta.getAllEnsembles().entrySet()) {
197-
if (curEntryId != null) {
198-
ledgerFragmentsRange.put(curEntryId, entry.getKey() - 1);
199-
}
200-
curEntryId = entry.getKey();
201-
}
202-
if (curEntryId != null) {
203-
ledgerFragmentsRange.put(curEntryId, lh.getLastAddConfirmed());
204-
}
205-
206-
for (Map.Entry<Long, ? extends List<BookieId>> entry : ledgerMeta.getAllEnsembles().entrySet()) {
207-
if (placementPolicy.isEnsembleAdheringToPlacementPolicy(entry.getValue(),
208-
ledgerMeta.getWriteQuorumSize(), ledgerMeta.getAckQuorumSize())
209-
== EnsemblePlacementPolicy.PlacementPolicyAdherence.FAIL) {
210-
try {
211-
final List<BookieId> currentEnsemble = entry.getValue();
212-
// Currently, don't consider quarantinedBookies
213-
final EnsemblePlacementPolicy.PlacementResult<List<BookieId>> placementResult =
214-
placementPolicy.replaceToAdherePlacementPolicy(
215-
ledgerMeta.getEnsembleSize(),
216-
ledgerMeta.getWriteQuorumSize(),
217-
ledgerMeta.getAckQuorumSize(),
218-
Collections.emptySet(),
219-
currentEnsemble);
220-
221-
if (placementResult.isAdheringToPolicy()
222-
== EnsemblePlacementPolicy.PlacementPolicyAdherence.FAIL) {
223-
LOG.warn("Failed to relocate the ensemble. So, skip the operation."
224-
+ " ledgerId: {}, fragmentIndex: {}",
225-
ledgerId, entry.getKey());
226-
failedTargets.add(Pair.of(ledgerId, entry.getKey()));
227-
} else {
228-
final List<BookieId> newEnsemble = placementResult.getResult();
229-
final Map<Integer, BookieId> replaceBookiesMap = IntStream
230-
.range(0, ledgerMeta.getEnsembleSize()).boxed()
231-
.filter(i -> !newEnsemble.get(i).equals(currentEnsemble.get(i)))
232-
.collect(Collectors.toMap(Functions.identity(), newEnsemble::get));
233-
if (replaceBookiesMap.isEmpty()) {
234-
LOG.warn("Failed to get bookies to replace. So, skip the operation."
235-
+ " ledgerId: {}, fragmentIndex: {}",
236-
ledgerId, entry.getKey());
237-
failedTargets.add(Pair.of(ledgerId, entry.getKey()));
238-
} else if (flags.dryRun) {
239-
LOG.info("Would replace the ensemble. ledgerId: {}, fragmentIndex: {},"
240-
+ " currentEnsemble: {} replaceBookiesMap {}",
241-
ledgerId, entry.getKey(),
242-
currentEnsemble, replaceBookiesMap);
243-
} else {
244-
if (LOG.isDebugEnabled()) {
245-
LOG.debug("Try to replace the ensemble. ledgerId: {}, fragmentIndex: {},"
246-
+ " replaceBookiesMap {}",
247-
ledgerId, entry.getKey(), replaceBookiesMap);
248-
}
249-
final LedgerFragment fragment = new LedgerFragment(lh, entry.getKey(),
250-
ledgerFragmentsRange.get(entry.getKey()), replaceBookiesMap.keySet());
251-
252-
try {
253-
admin.replicateLedgerFragment(lh, fragment, replaceBookiesMap,
254-
(lId, eId) -> {
255-
// This consumer is already accepted before the method returns
256-
// void. Therefore, use failedTargets in this consumer.
257-
LOG.warn("Failed to read entry {}:{}", lId, eId);
258-
failedTargets.add(Pair.of(ledgerId, entry.getKey()));
259-
});
260-
LOG.info("Operation finished in the ensemble. ledgerId: {},"
261-
+ " fragmentIndex: {}, replaceBookiesMap {}",
262-
ledgerId, entry.getKey(), replaceBookiesMap);
263-
} catch (BKException | InterruptedException e) {
264-
LOG.warn("Failed to replicate ledger fragment.", e);
265-
failedTargets.add(Pair.of(ledgerId, entry.getKey()));
266-
}
267-
}
268-
}
269-
} catch (UnsupportedOperationException e) {
270-
LOG.warn("UnsupportedOperationException caught. The placement policy might not support"
271-
+ " replaceToAdherePlacementPolicy method.", e);
272-
failedTargets.add(Pair.of(ledgerId, entry.getKey()));
273-
}
274-
} else {
275-
if (LOG.isDebugEnabled()) {
276-
LOG.debug("The fragment is adhering to placement policy. So, skip the operation."
277-
+ " ledgerId: {}, fragmentIndex: {}", ledgerId, entry.getKey());
278-
}
279-
}
280-
}
183+
admin.relocateLedgerToAdherePlacementPolicy(lh, flags.dryRun)
184+
.forEach(e -> failedTargets.add(Pair.of(ledgerId, e)));
185+
} catch (UnsupportedOperationException e) {
186+
LOG.warn("UnsupportedOperationException caught. The placement policy might not support"
187+
+ " replaceToAdherePlacementPolicy method.", e);
281188
} finally {
282189
try {
283190
if (!flags.dryRun) {
284191
lum.releaseUnderreplicatedLedger(ledgerId);
285192
}
286193
} catch (ReplicationException e) {
287-
LOG.warn("Failed to release under replicated ledger {}.", ledgerId, e);
194+
LOG.error("Failed to release under replicated ledger {}.", ledgerId, e);
288195
} finally {
289196
((CountDownLatch) ctx).countDown();
290197
}

0 commit comments

Comments
 (0)