Skip to content

Commit 52c6759

Browse files
KeeProMiseKeeProMise
committed
add atLeastWaitFor
1 parent fd13d38 commit 52c6759

File tree

2 files changed

+94
-53
lines changed

2 files changed

+94
-53
lines changed

‎hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -404,6 +404,28 @@ public static void waitFor(final Supplier<Boolean> check,
404404
}
405405
}
406406

407+
/**
408+
* Wait at least {@code atLeastWaitForMillis} from start to end of the test.
409+
*
410+
* @param check the test to perform.
411+
* @param atLeastWaitForMillis the minimum waiting time from the beginning
412+
* to the end of the test.
413+
* @throws InterruptedException if the method is interrupted while waiting.
414+
*/
415+
public static <T> T atLeastWaitFor(final Supplier<T> check,
416+
long atLeastWaitForMillis) throws InterruptedException {
417+
if (atLeastWaitForMillis < 0) {
418+
atLeastWaitForMillis = 0;
419+
}
420+
long st = Time.monotonicNow();
421+
T result = check.get();
422+
long runTime = Time.monotonicNow() - st;
423+
if (runTime < atLeastWaitForMillis) {
424+
Thread.sleep(atLeastWaitForMillis - runTime);
425+
}
426+
return result;
427+
}
428+
407429
/**
408430
* Prints output to one {@link PrintStream} while copying to the other.
409431
* <p>

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/TestNamenodeResolver.java‎

Lines changed: 72 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.getStateStoreConfiguration;
2727
import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.newStateStore;
2828
import static org.apache.hadoop.hdfs.server.federation.store.FederationStateStoreTestUtils.waitStateStore;
29+
import static org.apache.hadoop.test.GenericTestUtils.atLeastWaitFor;
2930
import static org.junit.Assert.assertEquals;
3031
import static org.junit.Assert.assertFalse;
3132
import static org.junit.Assert.assertNotNull;
@@ -282,13 +283,13 @@ public void testRegistrationNamenodeSelection()
282283
// 1) ns0:nn0 - Active
283284
// 2) ns0:nn1 - Standby (newest)
284285
// Verify the selected entry is the active entry
285-
assertTrue(namenodeResolver.registerNamenode(
286-
createNamenodeReport(
287-
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE)));
288-
Thread.sleep(100);
289-
assertTrue(namenodeResolver.registerNamenode(
290-
createNamenodeReport(
291-
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY)));
286+
assertTrue(atLeastWaitFor(
287+
() -> registerNamenode(
288+
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE),
289+
100));
290+
291+
assertTrue(registerNamenode(
292+
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY));
292293

293294
stateStore.refreshCaches(true);
294295

@@ -300,16 +301,15 @@ public void testRegistrationNamenodeSelection()
300301
// 2) ns0:nn1 - Standby (newest)
301302
// Verify the selected entry is the standby entry as the active entry is
302303
// stale
303-
assertTrue(namenodeResolver.registerNamenode(
304-
createNamenodeReport(
305-
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE)));
306-
307304
// Expire active registration
308-
Thread.sleep(6000);
305+
assertTrue(atLeastWaitFor(
306+
() -> registerNamenode(
307+
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE),
308+
6000));
309309

310310
// Refresh standby registration
311-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
312-
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY)));
311+
assertTrue(registerNamenode(
312+
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY));
313313

314314
// Verify that standby is selected (active is now expired)
315315
stateStore.refreshCaches(true);
@@ -319,23 +319,25 @@ public void testRegistrationNamenodeSelection()
319319
// 1) ns0:nn0 - Active
320320
// 2) ns0:nn1 - Unavailable (newest)
321321
// Verify the selected entry is the active entry
322-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
323-
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE)));
324-
Thread.sleep(100);
325-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
326-
NAMESERVICES[0], NAMENODES[1], null)));
322+
assertTrue(atLeastWaitFor(
323+
() -> registerNamenode(
324+
NAMESERVICES[0], NAMENODES[0], HAServiceState.ACTIVE),
325+
100));
326+
assertTrue(registerNamenode(
327+
NAMESERVICES[0], NAMENODES[1], null));
327328
stateStore.refreshCaches(true);
328329
verifyFirstRegistration(NAMESERVICES[0], NAMENODES[0], 2,
329330
FederationNamenodeServiceState.ACTIVE);
330331

331332
// 1) ns0:nn0 - Unavailable (newest)
332333
// 2) ns0:nn1 - Standby
333334
// Verify the selected entry is the standby entry
334-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
335-
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY)));
336-
Thread.sleep(1000);
337-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
338-
NAMESERVICES[0], NAMENODES[0], null)));
335+
assertTrue(atLeastWaitFor(
336+
() -> registerNamenode(
337+
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY),
338+
1000));
339+
assertTrue(registerNamenode(
340+
NAMESERVICES[0], NAMENODES[0], null));
339341

340342
stateStore.refreshCaches(true);
341343
verifyFirstRegistration(NAMESERVICES[0], NAMENODES[1], 2,
@@ -345,14 +347,16 @@ public void testRegistrationNamenodeSelection()
345347
// 2) ns0:nn1 - Standby
346348
// 3) ns0:nn2 - Active (newest)
347349
// Verify the selected entry is the newest active entry
348-
assertTrue(namenodeResolver.registerNamenode(
349-
createNamenodeReport(NAMESERVICES[0], NAMENODES[0], null)));
350-
Thread.sleep(100);
351-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
352-
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY)));
353-
Thread.sleep(100);
354-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
355-
NAMESERVICES[0], NAMENODES[2], HAServiceState.ACTIVE)));
350+
assertTrue(atLeastWaitFor(
351+
() -> registerNamenode(
352+
NAMESERVICES[0], NAMENODES[0], null),
353+
100));
354+
assertTrue(atLeastWaitFor(
355+
() -> registerNamenode(
356+
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY),
357+
100));
358+
assertTrue(registerNamenode(
359+
NAMESERVICES[0], NAMENODES[2], HAServiceState.ACTIVE));
356360

357361
stateStore.refreshCaches(true);
358362
verifyFirstRegistration(NAMESERVICES[0], NAMENODES[2], 3,
@@ -362,14 +366,16 @@ public void testRegistrationNamenodeSelection()
362366
// 2) ns0:nn1 - Standby
363367
// 3) ns0:nn2 - Standby (newest)
364368
// Verify the selected entry is the oldest standby entry
365-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
366-
NAMESERVICES[0], NAMENODES[0], HAServiceState.STANDBY)));
367-
Thread.sleep(1500);
368-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
369-
NAMESERVICES[0], NAMENODES[2], HAServiceState.STANDBY)));
370-
Thread.sleep(1500);
371-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
372-
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY)));
369+
assertTrue(atLeastWaitFor(
370+
() -> registerNamenode(
371+
NAMESERVICES[0], NAMENODES[0], HAServiceState.STANDBY),
372+
1500));
373+
assertTrue(atLeastWaitFor(
374+
() -> registerNamenode(
375+
NAMESERVICES[0], NAMENODES[2], HAServiceState.STANDBY),
376+
1500));
377+
assertTrue(registerNamenode(
378+
NAMESERVICES[0], NAMENODES[1], HAServiceState.STANDBY));
373379

374380
stateStore.refreshCaches(true);
375381
verifyFirstRegistration(NAMESERVICES[0], NAMENODES[0], 3,
@@ -378,11 +384,12 @@ public void testRegistrationNamenodeSelection()
378384
// 1) ns0:nn0 - Observer (oldest)
379385
// 2) ns0:nn1 - Observer (newest)
380386
// Verify the selected entry is the newest Observer entry
381-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
382-
NAMESERVICES[0], NAMENODES[0], HAServiceState.OBSERVER)));
383-
Thread.sleep(1500);
384-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
385-
NAMESERVICES[0], NAMENODES[1], HAServiceState.OBSERVER)));
387+
assertTrue(atLeastWaitFor(
388+
() -> registerNamenode(
389+
NAMESERVICES[0], NAMENODES[0], HAServiceState.OBSERVER),
390+
1500));
391+
assertTrue(registerNamenode(
392+
NAMESERVICES[0], NAMENODES[1], HAServiceState.OBSERVER));
386393

387394
stateStore.refreshCaches(true);
388395
verifyFirstRegistration(NAMESERVICES[0], NAMENODES[1], 3,
@@ -392,14 +399,16 @@ public void testRegistrationNamenodeSelection()
392399
// 2) ns0:nn1 - Unavailable
393400
// 3) ns0:nn2 - Unavailable (newest)
394401
// Verify the selected entry is the oldest Unavailable entry
395-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
396-
NAMESERVICES[0], NAMENODES[0], null)));
397-
Thread.sleep(1500);
398-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
399-
NAMESERVICES[0], NAMENODES[1], null)));
400-
Thread.sleep(1500);
401-
assertTrue(namenodeResolver.registerNamenode(createNamenodeReport(
402-
NAMESERVICES[0], NAMENODES[2], null)));
402+
assertTrue(atLeastWaitFor(
403+
() -> registerNamenode(
404+
NAMESERVICES[0], NAMENODES[0], null),
405+
1500));
406+
assertTrue(atLeastWaitFor(
407+
() -> registerNamenode(
408+
NAMESERVICES[0], NAMENODES[1], null),
409+
1500));
410+
assertTrue(registerNamenode(
411+
NAMESERVICES[0], NAMENODES[2], null));
403412
stateStore.refreshCaches(true);
404413
verifyFirstRegistration(NAMESERVICES[0], NAMENODES[0], 3,
405414
FederationNamenodeServiceState.UNAVAILABLE);
@@ -457,4 +466,14 @@ private static InetSocketAddress getInetSocketAddress(String rpcAddr) {
457466
String hostname = rpcAddrArr[0];
458467
return new InetSocketAddress(hostname, port);
459468
}
469+
470+
private boolean registerNamenode(String nsId,
471+
String nnId, HAServiceState haServiceState) {
472+
try {
473+
return namenodeResolver.registerNamenode(
474+
createNamenodeReport(nsId, nnId, haServiceState));
475+
}catch (IOException e) {
476+
return false;
477+
}
478+
}
460479
}

0 commit comments

Comments
 (0)