Skip to content

Commit 208d4bd

Browse files
rhalmrhalm
authored
HDDS-11063. TestSnapshotDiffManager#testThreadPoolIsFull is flaky without wait between batches (#10581)
1 parent d4c06b2 commit 208d4bd

3 files changed

Lines changed: 88 additions & 87 deletions

File tree

‎hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,6 @@
5050
import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_NOT_FOUND;
5151
import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_KEY_NAME;
5252
import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.TIMEOUT;
53-
import static org.apache.hadoop.ozone.om.snapshot.SnapshotDiffManager.getSnapshotRootPath;
5453
import static org.apache.hadoop.ozone.om.snapshot.SnapshotUtils.checkSnapshotActive;
5554
import static org.apache.hadoop.ozone.om.snapshot.SnapshotUtils.dropColumnFamilyHandle;
5655
import static org.apache.hadoop.ozone.om.snapshot.db.SnapshotDiffDBDefinition.SNAP_DIFF_PURGED_JOB_TABLE_NAME;
@@ -842,7 +841,7 @@ public SnapshotDiffResponse getSnapshotDiffReport(final String volume,
842841
// Check if fromSnapshot and toSnapshot are equal.
843842
if (Objects.equals(fromSnapshot, toSnapshot)) {
844843
SnapshotDiffReportOzone diffReport = new SnapshotDiffReportOzone(
845-
getSnapshotRootPath(volume, bucket).toString(), volume, bucket,
844+
snapshotDiffManager.getSnapshotRootPath(volume, bucket).toString(), volume, bucket,
846845
fromSnapshot, toSnapshot, Collections.emptyList(), null);
847846
return new SnapshotDiffResponse(diffReport, DONE, 0L);
848847
}
@@ -873,7 +872,7 @@ public SnapshotDiffResponse getSnapshotDiffResponse(final String volume,
873872
// Check if fromSnapshot and toSnapshot are equal.
874873
if (Objects.equals(fromSnapshot, toSnapshot)) {
875874
SnapshotDiffReportOzone diffReport = new SnapshotDiffReportOzone(
876-
getSnapshotRootPath(volume, bucket).toString(), volume, bucket,
875+
snapshotDiffManager.getSnapshotRootPath(volume, bucket).toString(), volume, bucket,
877876
fromSnapshot, toSnapshot, Collections.emptyList(), null);
878877
return new SnapshotDiffResponse(diffReport, DONE, 0L);
879878
}

‎hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -692,10 +692,10 @@ public synchronized SubmitSnapshotDiffResponse submitSnapshotDiff(
692692
}
693693

694694
@Nonnull
695-
public static OFSPath getSnapshotRootPath(String volume, String bucket) {
695+
public OFSPath getSnapshotRootPath(String volume, String bucket) {
696696
org.apache.hadoop.fs.Path bucketPath = new org.apache.hadoop.fs.Path(
697697
OZONE_URI_DELIMITER + volume + OZONE_URI_DELIMITER + bucket);
698-
return new OFSPath(bucketPath, new OzoneConfiguration());
698+
return new OFSPath(bucketPath, ozoneManager.getConfiguration());
699699
}
700700

701701
@VisibleForTesting

‎hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java‎

Lines changed: 84 additions & 82 deletions
Original file line numberDiff line numberDiff line change
@@ -95,11 +95,8 @@
9595
import java.util.Optional;
9696
import java.util.Set;
9797
import java.util.UUID;
98+
import java.util.concurrent.CountDownLatch;
9899
import java.util.concurrent.ExecutionException;
99-
import java.util.concurrent.ExecutorService;
100-
import java.util.concurrent.Future;
101-
import java.util.concurrent.SynchronousQueue;
102-
import java.util.concurrent.ThreadPoolExecutor;
103100
import java.util.concurrent.TimeUnit;
104101
import java.util.concurrent.atomic.AtomicInteger;
105102
import java.util.function.BiFunction;
@@ -1169,106 +1166,111 @@ private void uploadSnapshotDiffJobToDb(SnapshotInfo fromSnapshot,
11691166
}
11701167

11711168
private static Stream<Arguments> threadPoolFullScenarios() {
1169+
int fullThreadPoolSize = 2 * OZONE_OM_SNAPSHOT_DIFF_THREAD_POOL_SIZE_DEFAULT;
11721170
return Stream.of(
1173-
Arguments.of("When there is a wait time between job batches",
1174-
500L, 45, 0),
1175-
Arguments.of("When there is no wait time between job batches",
1176-
0L, 20, 25)
1171+
Arguments.of("When the pool drains between job batches",
1172+
true, 45, 0),
1173+
Arguments.of("When the pool does not drain between job batches",
1174+
false, fullThreadPoolSize, 45 - fullThreadPoolSize)
11771175
);
11781176
}
11791177

11801178
@ParameterizedTest(name = "{0}")
11811179
@MethodSource("threadPoolFullScenarios")
11821180
public void testThreadPoolIsFull(String description,
1183-
long waitBetweenBatches,
1181+
boolean drainBetweenBatches,
11841182
int expectInProgressJobsCount,
11851183
int expectRejectedJobsCount)
11861184
throws Exception {
1187-
ExecutorService executorService = new ThreadPoolExecutor(100, 100, 0,
1188-
TimeUnit.MILLISECONDS, new SynchronousQueue<>()
1189-
);
1190-
1191-
List<SnapshotInfo> snapshotInfos = new ArrayList<>();
1192-
1193-
for (int i = 0; i < 10; i++) {
1194-
UUID snapshotId = UUID.randomUUID();
1195-
String snapshotName = "snap-" + snapshotId;
1196-
SnapshotInfo snapInfo = new SnapshotInfo.Builder()
1197-
.setSnapshotId(snapshotId)
1198-
.setVolumeName(VOLUME_NAME)
1199-
.setBucketName(BUCKET_NAME)
1200-
.setName(snapshotName)
1201-
.setSnapshotPath("fromSnapshotPath")
1202-
.build();
1203-
snapshotInfos.add(snapInfo);
1204-
1205-
when(snapshotInfoTable.get(getTableKey(VOLUME_NAME, BUCKET_NAME,
1206-
snapshotName))).thenReturn(snapInfo);
1207-
}
1208-
1185+
List<SnapshotInfo> snapshotInfos = createTestSnapshots(10);
12091186
SnapshotDiffManager spy = spy(snapshotDiffManager);
12101187

1211-
for (int i = 0; i < snapshotInfos.size(); i++) {
1212-
for (int j = i + 1; j < snapshotInfos.size(); j++) {
1213-
String fromSnapshotName = snapshotInfos.get(i).getName();
1214-
String toSnapshotName = snapshotInfos.get(j).getName();
1188+
CountDownLatch blockWorkers = new CountDownLatch(1);
1189+
AtomicInteger completedJobs = new AtomicInteger(0);
1190+
doAnswer(invocation -> {
1191+
blockWorkers.await();
1192+
completedJobs.incrementAndGet();
1193+
return null;
1194+
}).when(spy).generateSnapshotDiffReport(anyString(), anyString(),
1195+
eq(VOLUME_NAME), eq(BUCKET_NAME), anyString(), anyString(),
1196+
eq(false), eq(false));
12151197

1216-
doAnswer(invocation -> {
1217-
Thread.sleep(250L);
1218-
return null;
1219-
}).when(spy).generateSnapshotDiffReport(anyString(), anyString(),
1220-
eq(VOLUME_NAME), eq(BUCKET_NAME), eq(fromSnapshotName),
1221-
eq(toSnapshotName), eq(false), eq(false));
1222-
}
1198+
if (drainBetweenBatches) {
1199+
blockWorkers.countDown();
12231200
}
12241201

1225-
List<Future<SnapshotDiffResponse>> futures = new ArrayList<>();
1226-
for (int i = 0; i < snapshotInfos.size(); i++) {
1227-
for (int j = i + 1; j < snapshotInfos.size(); j++) {
1228-
String fromSnapshotName = snapshotInfos.get(i).getName();
1229-
String toSnapshotName = snapshotInfos.get(j).getName();
1230-
1231-
Future<SnapshotDiffResponse> future = executorService.submit(
1232-
() -> submitJob(spy, fromSnapshotName, toSnapshotName));
1233-
futures.add(future);
1202+
try {
1203+
List<SnapshotDiffResponse> responses = new ArrayList<>();
1204+
int totalSubmitted = 0;
1205+
for (int i = 0; i < snapshotInfos.size(); i++) {
1206+
for (int j = i + 1; j < snapshotInfos.size(); j++) {
1207+
String fromSnapshotName = snapshotInfos.get(i).getName();
1208+
String toSnapshotName = snapshotInfos.get(j).getName();
1209+
responses.add(submitJob(spy, fromSnapshotName, toSnapshotName));
1210+
totalSubmitted++;
1211+
}
1212+
if (drainBetweenBatches) {
1213+
final int expected = totalSubmitted;
1214+
attempt(() -> {
1215+
if (completedJobs.get() < expected) {
1216+
throw new IllegalStateException("Waiting for jobs to complete");
1217+
}
1218+
return null;
1219+
}, 50, TimeDuration.valueOf(100, TimeUnit.MILLISECONDS), null, null);
1220+
}
12341221
}
1235-
Thread.sleep(waitBetweenBatches);
1236-
}
12371222

1238-
// Wait to make sure that all jobs finish before assertion.
1239-
Thread.sleep(1000L);
1240-
int inProgressJobsCount = 0;
1241-
int rejectedJobsCount = 0;
1242-
1243-
for (Future<SnapshotDiffResponse> future : futures) {
1244-
SnapshotDiffResponse response = future.get();
1245-
if (response.getJobStatus() == IN_PROGRESS) {
1246-
inProgressJobsCount++;
1247-
} else if (response.getJobStatus() == REJECTED) {
1248-
rejectedJobsCount++;
1249-
} else {
1250-
throw new IllegalStateException("Unexpected job status.");
1223+
int inProgressJobsCount = 0;
1224+
int rejectedJobsCount = 0;
1225+
for (SnapshotDiffResponse response : responses) {
1226+
if (response.getJobStatus() == IN_PROGRESS) {
1227+
inProgressJobsCount++;
1228+
} else if (response.getJobStatus() == REJECTED) {
1229+
rejectedJobsCount++;
1230+
} else {
1231+
throw new IllegalStateException("Unexpected job status.");
1232+
}
12511233
}
1252-
}
12531234

1254-
assertEquals(expectInProgressJobsCount, inProgressJobsCount);
1255-
assertEquals(expectRejectedJobsCount, rejectedJobsCount);
1256-
1257-
int notFoundJobs = 0;
1258-
for (int i = 0; i < snapshotInfos.size(); i++) {
1259-
for (int j = i + 1; j < snapshotInfos.size(); j++) {
1260-
SnapshotDiffJob diffJob =
1261-
getSnapshotDiffJobFromDb(snapshotInfos.get(i),
1262-
snapshotInfos.get(j));
1263-
if (diffJob == null) {
1264-
notFoundJobs++;
1235+
assertEquals(expectInProgressJobsCount, inProgressJobsCount);
1236+
assertEquals(expectRejectedJobsCount, rejectedJobsCount);
1237+
1238+
int notFoundJobs = 0;
1239+
for (int i = 0; i < snapshotInfos.size(); i++) {
1240+
for (int j = i + 1; j < snapshotInfos.size(); j++) {
1241+
SnapshotDiffJob diffJob =
1242+
getSnapshotDiffJobFromDb(snapshotInfos.get(i),
1243+
snapshotInfos.get(j));
1244+
if (diffJob == null) {
1245+
notFoundJobs++;
1246+
}
12651247
}
12661248
}
1249+
1250+
// assert that rejected jobs were removed from the job table as well.
1251+
assertEquals(expectRejectedJobsCount, notFoundJobs);
1252+
} finally {
1253+
blockWorkers.countDown();
12671254
}
1255+
}
12681256

1269-
// assert that rejected jobs were removed from the job table as well.
1270-
assertEquals(expectRejectedJobsCount, notFoundJobs);
1271-
executorService.shutdown();
1257+
private List<SnapshotInfo> createTestSnapshots(int count) throws IOException {
1258+
List<SnapshotInfo> snapshotInfos = new ArrayList<>();
1259+
for (int i = 0; i < count; i++) {
1260+
UUID snapshotId = UUID.randomUUID();
1261+
String snapshotName = "snap-" + snapshotId;
1262+
SnapshotInfo snapInfo = new SnapshotInfo.Builder()
1263+
.setSnapshotId(snapshotId)
1264+
.setVolumeName(VOLUME_NAME)
1265+
.setBucketName(BUCKET_NAME)
1266+
.setName(snapshotName)
1267+
.setSnapshotPath("fromSnapshotPath")
1268+
.build();
1269+
snapshotInfos.add(snapInfo);
1270+
when(snapshotInfoTable.get(getTableKey(VOLUME_NAME, BUCKET_NAME, snapshotName)))
1271+
.thenReturn(snapInfo);
1272+
}
1273+
return snapshotInfos;
12721274
}
12731275

12741276
private SnapshotDiffResponse submitJob(SnapshotDiffManager diffManager,
@@ -1641,7 +1643,7 @@ public void testGetSnapshotDiffReportWhenDone() throws Exception {
16411643

16421644
SnapshotDiffManager spy = spy(snapshotDiffManager);
16431645
SnapshotDiffReportOzone dummyReport = new SnapshotDiffReportOzone(
1644-
SnapshotDiffManager.getSnapshotRootPath(ctx.volumeName, ctx.bucketName).toString(),
1646+
spy.getSnapshotRootPath(ctx.volumeName, ctx.bucketName).toString(),
16451647
ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName, ctx.toSnapshotName,
16461648
expectedEntries, null);
16471649
doReturn(dummyReport).when(spy).createPageResponse(any(SnapshotDiffJob.class),

0 commit comments

Comments
 (0)