This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 208d4bd8b56 HDDS-11063. TestSnapshotDiffManager#testThreadPoolIsFull
is flaky without wait between batches (#10581)
208d4bd8b56 is described below
commit 208d4bd8b5668aa2b33098e6b500c116ad84fa86
Author: rhalm <[email protected]>
AuthorDate: Wed Jun 24 07:20:10 2026 +0200
HDDS-11063. TestSnapshotDiffManager#testThreadPoolIsFull is flaky without
wait between batches (#10581)
---
.../apache/hadoop/ozone/om/OmSnapshotManager.java | 5 +-
.../ozone/om/snapshot/SnapshotDiffManager.java | 4 +-
.../ozone/om/snapshot/TestSnapshotDiffManager.java | 166 +++++++++++----------
3 files changed, 88 insertions(+), 87 deletions(-)
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java
index 870240c36d3..7658d026452 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java
@@ -50,7 +50,6 @@
import static
org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_NOT_FOUND;
import static
org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_KEY_NAME;
import static
org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.TIMEOUT;
-import static
org.apache.hadoop.ozone.om.snapshot.SnapshotDiffManager.getSnapshotRootPath;
import static
org.apache.hadoop.ozone.om.snapshot.SnapshotUtils.checkSnapshotActive;
import static
org.apache.hadoop.ozone.om.snapshot.SnapshotUtils.dropColumnFamilyHandle;
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,
// Check if fromSnapshot and toSnapshot are equal.
if (Objects.equals(fromSnapshot, toSnapshot)) {
SnapshotDiffReportOzone diffReport = new SnapshotDiffReportOzone(
- getSnapshotRootPath(volume, bucket).toString(), volume, bucket,
+ snapshotDiffManager.getSnapshotRootPath(volume, bucket).toString(),
volume, bucket,
fromSnapshot, toSnapshot, Collections.emptyList(), null);
return new SnapshotDiffResponse(diffReport, DONE, 0L);
}
@@ -873,7 +872,7 @@ public SnapshotDiffResponse getSnapshotDiffResponse(final
String volume,
// Check if fromSnapshot and toSnapshot are equal.
if (Objects.equals(fromSnapshot, toSnapshot)) {
SnapshotDiffReportOzone diffReport = new SnapshotDiffReportOzone(
- getSnapshotRootPath(volume, bucket).toString(), volume, bucket,
+ snapshotDiffManager.getSnapshotRootPath(volume, bucket).toString(),
volume, bucket,
fromSnapshot, toSnapshot, Collections.emptyList(), null);
return new SnapshotDiffResponse(diffReport, DONE, 0L);
}
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java
index 418650578ff..e885e2b4f68 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java
@@ -692,10 +692,10 @@ public synchronized SubmitSnapshotDiffResponse
submitSnapshotDiff(
}
@Nonnull
- public static OFSPath getSnapshotRootPath(String volume, String bucket) {
+ public OFSPath getSnapshotRootPath(String volume, String bucket) {
org.apache.hadoop.fs.Path bucketPath = new org.apache.hadoop.fs.Path(
OZONE_URI_DELIMITER + volume + OZONE_URI_DELIMITER + bucket);
- return new OFSPath(bucketPath, new OzoneConfiguration());
+ return new OFSPath(bucketPath, ozoneManager.getConfiguration());
}
@VisibleForTesting
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java
index ddebf54d600..5b1039b165d 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java
@@ -95,11 +95,8 @@
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
@@ -1169,106 +1166,111 @@ private void uploadSnapshotDiffJobToDb(SnapshotInfo
fromSnapshot,
}
private static Stream<Arguments> threadPoolFullScenarios() {
+ int fullThreadPoolSize = 2 *
OZONE_OM_SNAPSHOT_DIFF_THREAD_POOL_SIZE_DEFAULT;
return Stream.of(
- Arguments.of("When there is a wait time between job batches",
- 500L, 45, 0),
- Arguments.of("When there is no wait time between job batches",
- 0L, 20, 25)
+ Arguments.of("When the pool drains between job batches",
+ true, 45, 0),
+ Arguments.of("When the pool does not drain between job batches",
+ false, fullThreadPoolSize, 45 - fullThreadPoolSize)
);
}
@ParameterizedTest(name = "{0}")
@MethodSource("threadPoolFullScenarios")
public void testThreadPoolIsFull(String description,
- long waitBetweenBatches,
+ boolean drainBetweenBatches,
int expectInProgressJobsCount,
int expectRejectedJobsCount)
throws Exception {
- ExecutorService executorService = new ThreadPoolExecutor(100, 100, 0,
- TimeUnit.MILLISECONDS, new SynchronousQueue<>()
- );
-
- List<SnapshotInfo> snapshotInfos = new ArrayList<>();
-
- for (int i = 0; i < 10; i++) {
- UUID snapshotId = UUID.randomUUID();
- String snapshotName = "snap-" + snapshotId;
- SnapshotInfo snapInfo = new SnapshotInfo.Builder()
- .setSnapshotId(snapshotId)
- .setVolumeName(VOLUME_NAME)
- .setBucketName(BUCKET_NAME)
- .setName(snapshotName)
- .setSnapshotPath("fromSnapshotPath")
- .build();
- snapshotInfos.add(snapInfo);
-
- when(snapshotInfoTable.get(getTableKey(VOLUME_NAME, BUCKET_NAME,
- snapshotName))).thenReturn(snapInfo);
- }
-
+ List<SnapshotInfo> snapshotInfos = createTestSnapshots(10);
SnapshotDiffManager spy = spy(snapshotDiffManager);
- for (int i = 0; i < snapshotInfos.size(); i++) {
- for (int j = i + 1; j < snapshotInfos.size(); j++) {
- String fromSnapshotName = snapshotInfos.get(i).getName();
- String toSnapshotName = snapshotInfos.get(j).getName();
+ CountDownLatch blockWorkers = new CountDownLatch(1);
+ AtomicInteger completedJobs = new AtomicInteger(0);
+ doAnswer(invocation -> {
+ blockWorkers.await();
+ completedJobs.incrementAndGet();
+ return null;
+ }).when(spy).generateSnapshotDiffReport(anyString(), anyString(),
+ eq(VOLUME_NAME), eq(BUCKET_NAME), anyString(), anyString(),
+ eq(false), eq(false));
- doAnswer(invocation -> {
- Thread.sleep(250L);
- return null;
- }).when(spy).generateSnapshotDiffReport(anyString(), anyString(),
- eq(VOLUME_NAME), eq(BUCKET_NAME), eq(fromSnapshotName),
- eq(toSnapshotName), eq(false), eq(false));
- }
+ if (drainBetweenBatches) {
+ blockWorkers.countDown();
}
- List<Future<SnapshotDiffResponse>> futures = new ArrayList<>();
- for (int i = 0; i < snapshotInfos.size(); i++) {
- for (int j = i + 1; j < snapshotInfos.size(); j++) {
- String fromSnapshotName = snapshotInfos.get(i).getName();
- String toSnapshotName = snapshotInfos.get(j).getName();
-
- Future<SnapshotDiffResponse> future = executorService.submit(
- () -> submitJob(spy, fromSnapshotName, toSnapshotName));
- futures.add(future);
+ try {
+ List<SnapshotDiffResponse> responses = new ArrayList<>();
+ int totalSubmitted = 0;
+ for (int i = 0; i < snapshotInfos.size(); i++) {
+ for (int j = i + 1; j < snapshotInfos.size(); j++) {
+ String fromSnapshotName = snapshotInfos.get(i).getName();
+ String toSnapshotName = snapshotInfos.get(j).getName();
+ responses.add(submitJob(spy, fromSnapshotName, toSnapshotName));
+ totalSubmitted++;
+ }
+ if (drainBetweenBatches) {
+ final int expected = totalSubmitted;
+ attempt(() -> {
+ if (completedJobs.get() < expected) {
+ throw new IllegalStateException("Waiting for jobs to complete");
+ }
+ return null;
+ }, 50, TimeDuration.valueOf(100, TimeUnit.MILLISECONDS), null, null);
+ }
}
- Thread.sleep(waitBetweenBatches);
- }
- // Wait to make sure that all jobs finish before assertion.
- Thread.sleep(1000L);
- int inProgressJobsCount = 0;
- int rejectedJobsCount = 0;
-
- for (Future<SnapshotDiffResponse> future : futures) {
- SnapshotDiffResponse response = future.get();
- if (response.getJobStatus() == IN_PROGRESS) {
- inProgressJobsCount++;
- } else if (response.getJobStatus() == REJECTED) {
- rejectedJobsCount++;
- } else {
- throw new IllegalStateException("Unexpected job status.");
+ int inProgressJobsCount = 0;
+ int rejectedJobsCount = 0;
+ for (SnapshotDiffResponse response : responses) {
+ if (response.getJobStatus() == IN_PROGRESS) {
+ inProgressJobsCount++;
+ } else if (response.getJobStatus() == REJECTED) {
+ rejectedJobsCount++;
+ } else {
+ throw new IllegalStateException("Unexpected job status.");
+ }
}
- }
- assertEquals(expectInProgressJobsCount, inProgressJobsCount);
- assertEquals(expectRejectedJobsCount, rejectedJobsCount);
-
- int notFoundJobs = 0;
- for (int i = 0; i < snapshotInfos.size(); i++) {
- for (int j = i + 1; j < snapshotInfos.size(); j++) {
- SnapshotDiffJob diffJob =
- getSnapshotDiffJobFromDb(snapshotInfos.get(i),
- snapshotInfos.get(j));
- if (diffJob == null) {
- notFoundJobs++;
+ assertEquals(expectInProgressJobsCount, inProgressJobsCount);
+ assertEquals(expectRejectedJobsCount, rejectedJobsCount);
+
+ int notFoundJobs = 0;
+ for (int i = 0; i < snapshotInfos.size(); i++) {
+ for (int j = i + 1; j < snapshotInfos.size(); j++) {
+ SnapshotDiffJob diffJob =
+ getSnapshotDiffJobFromDb(snapshotInfos.get(i),
+ snapshotInfos.get(j));
+ if (diffJob == null) {
+ notFoundJobs++;
+ }
}
}
+
+ // assert that rejected jobs were removed from the job table as well.
+ assertEquals(expectRejectedJobsCount, notFoundJobs);
+ } finally {
+ blockWorkers.countDown();
}
+ }
- // assert that rejected jobs were removed from the job table as well.
- assertEquals(expectRejectedJobsCount, notFoundJobs);
- executorService.shutdown();
+ private List<SnapshotInfo> createTestSnapshots(int count) throws IOException
{
+ List<SnapshotInfo> snapshotInfos = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ UUID snapshotId = UUID.randomUUID();
+ String snapshotName = "snap-" + snapshotId;
+ SnapshotInfo snapInfo = new SnapshotInfo.Builder()
+ .setSnapshotId(snapshotId)
+ .setVolumeName(VOLUME_NAME)
+ .setBucketName(BUCKET_NAME)
+ .setName(snapshotName)
+ .setSnapshotPath("fromSnapshotPath")
+ .build();
+ snapshotInfos.add(snapInfo);
+ when(snapshotInfoTable.get(getTableKey(VOLUME_NAME, BUCKET_NAME,
snapshotName)))
+ .thenReturn(snapInfo);
+ }
+ return snapshotInfos;
}
private SnapshotDiffResponse submitJob(SnapshotDiffManager diffManager,
@@ -1641,7 +1643,7 @@ public void testGetSnapshotDiffReportWhenDone() throws
Exception {
SnapshotDiffManager spy = spy(snapshotDiffManager);
SnapshotDiffReportOzone dummyReport = new SnapshotDiffReportOzone(
- SnapshotDiffManager.getSnapshotRootPath(ctx.volumeName,
ctx.bucketName).toString(),
+ spy.getSnapshotRootPath(ctx.volumeName, ctx.bucketName).toString(),
ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName,
ctx.toSnapshotName,
expectedEntries, null);
doReturn(dummyReport).when(spy).createPageResponse(any(SnapshotDiffJob.class),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
