Skip to content

Commit 0376bdc

Browse files
authored
Optimize reorderReadSequence to check WriteSet instead of entire Ensemble (#4478)
1 parent f148f63 commit 0376bdc

File tree

2 files changed

+42
-2
lines changed

2 files changed

+42
-2
lines changed

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/RackawareEnsemblePlacementPolicyImpl.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -920,8 +920,9 @@ DistributionSchedule.WriteSet reorderReadSequenceWithRegion(
920920
if (useRegionAware || reorderReadsRandom) {
921921
isAnyBookieUnavailable = true;
922922
} else {
923-
for (int i = 0; i < ensemble.size(); i++) {
924-
BookieId bookieAddr = ensemble.get(i);
923+
for (int i = 0; i < writeSet.size(); i++) {
924+
int idx = writeSet.get(i);
925+
BookieId bookieAddr = ensemble.get(idx);
925926
if ((!knownBookies.containsKey(bookieAddr) && !readOnlyBookies.contains(bookieAddr))
926927
|| slowBookies.getIfPresent(bookieAddr) != null) {
927928
// Found at least one bookie not available in the ensemble, or in slowBookies

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestRackawareEnsemblePlacementPolicy.java

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
import org.apache.bookkeeper.stats.Gauge;
6565
import org.apache.bookkeeper.stats.NullStatsLogger;
6666
import org.apache.bookkeeper.test.TestStatsProvider;
67+
import org.apache.bookkeeper.test.TestStatsProvider.TestOpStatsLogger;
6768
import org.apache.bookkeeper.test.TestStatsProvider.TestStatsLogger;
6869
import org.apache.bookkeeper.util.StaticDNSResolver;
6970
import org.apache.commons.collections4.CollectionUtils;
@@ -2359,6 +2360,44 @@ public void testNodeWithFailures() throws Exception {
23592360
StaticDNSResolver.reset();
23602361
}
23612362

2363+
@Test
2364+
public void testSlowBookieInEnsembleOnly() throws Exception {
2365+
repp.uninitalize();
2366+
updateMyRack("/r1/rack1");
2367+
2368+
TestStatsProvider statsProvider = new TestStatsProvider();
2369+
TestStatsLogger statsLogger = statsProvider.getStatsLogger("");
2370+
2371+
repp = new RackawareEnsemblePlacementPolicy();
2372+
repp.initialize(conf, Optional.<DNSToSwitchMapping>empty(), timer,
2373+
DISABLE_ALL, statsLogger, BookieSocketAddress.LEGACY_BOOKIEID_RESOLVER);
2374+
repp.withDefaultRack(NetworkTopology.DEFAULT_REGION_AND_RACK);
2375+
2376+
TestOpStatsLogger readRequestsReorderedCounter = (TestOpStatsLogger) statsLogger
2377+
.getOpStatsLogger(BookKeeperClientStats.READ_REQUESTS_REORDERED);
2378+
2379+
// Update cluster
2380+
Set<BookieId> addrs = new HashSet<BookieId>();
2381+
addrs.add(addr1.toBookieId());
2382+
addrs.add(addr2.toBookieId());
2383+
addrs.add(addr3.toBookieId());
2384+
addrs.add(addr4.toBookieId());
2385+
repp.onClusterChanged(addrs, new HashSet<BookieId>());
2386+
repp.registerSlowBookie(addr1.toBookieId(), 0L);
2387+
Map<BookieId, Long> bookiePendingMap = new HashMap<>();
2388+
bookiePendingMap.put(addr1.toBookieId(), 1L);
2389+
repp.onClusterChanged(addrs, new HashSet<>());
2390+
2391+
DistributionSchedule.WriteSet writeSet = writeSetFromValues(1, 2, 3);
2392+
2393+
DistributionSchedule.WriteSet reorderSet = repp.reorderReadSequence(
2394+
ensemble, getBookiesHealthInfo(new HashMap<>(), bookiePendingMap), writeSet);
2395+
2396+
// If the slow bookie is only present in the ensemble, no reordering occurs.
2397+
assertEquals(writeSet, reorderSet);
2398+
assertEquals(0, readRequestsReorderedCounter.getSuccessCount());
2399+
}
2400+
23622401
@Test
23632402
public void testReplaceNotAvailableBookieWithDefaultRack() throws Exception {
23642403
repp.uninitalize();

0 commit comments

Comments
 (0)