From 38925a67766f163f6e7fb89676c0b11a457ff110 Mon Sep 17 00:00:00 2001 From: Renata Halmschlager <49129049+rhalm@users.noreply.github.com> Date: Mon, 22 Jun 2026 15:37:21 +0200 Subject: [PATCH 1/2] HDDS-11063. TestSnapshotDiffManager#testThreadPoolIsFull is flaky when there is no wait between the batches --- .../hadoop/ozone/om/OmSnapshotManager.java | 5 +- .../om/snapshot/SnapshotDiffManager.java | 4 +- .../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 870240c36d3a..7658d0264521 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 418650578ffa..e885e2b4f689 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 ddebf54d6006..e889ebb56200 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; @@ -1170,105 +1167,110 @@ private void uploadSnapshotDiffJobToDb(SnapshotInfo fromSnapshot, private static Stream threadPoolFullScenarios() { 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), + // 10 running + 10 queued = 20 accepted, remaining 25 rejected + Arguments.of("When the pool does not drain between job batches", + false, 20, 25) ); } @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 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 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> 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 future = executorService.submit( - () -> submitJob(spy, fromSnapshotName, toSnapshotName)); - futures.add(future); + try { + List 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 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 createTestSnapshots(int count) throws IOException { + List 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), From 7a91eb4a75bc36491d28276dd1b2005d8265feb3 Mon Sep 17 00:00:00 2001 From: Renata Halmschlager <49129049+rhalm@users.noreply.github.com> Date: Tue, 23 Jun 2026 13:56:14 +0200 Subject: [PATCH 2/2] HDDS-11063. Use full thread pool size to derive expected job counts --- .../hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 e889ebb56200..5b1039b165d6 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 @@ -1166,12 +1166,12 @@ private void uploadSnapshotDiffJobToDb(SnapshotInfo fromSnapshot, } private static Stream threadPoolFullScenarios() { + int fullThreadPoolSize = 2 * OZONE_OM_SNAPSHOT_DIFF_THREAD_POOL_SIZE_DEFAULT; return Stream.of( Arguments.of("When the pool drains between job batches", true, 45, 0), - // 10 running + 10 queued = 20 accepted, remaining 25 rejected Arguments.of("When the pool does not drain between job batches", - false, 20, 25) + false, fullThreadPoolSize, 45 - fullThreadPoolSize) ); }