From e003eeb25601f7adac6d167474efa5244abe50bc Mon Sep 17 00:00:00 2001 From: Wei-Chiu Chuang Date: Tue, 27 Oct 2026 16:10:22 -0700 Subject: [PATCH 1/3] HDDS-16442. Fix snapshot diff integrity failure recovery after upgrade Change-Id: I27661c03858cc1e82677ef502d26b6bc44f7d0d2 --- .../om/snapshot/SnapshotDiffManager.java | 36 ++++--- .../om/snapshot/TestSnapshotDiffManager.java | 93 +++++++++++++++++-- 2 files changed, 112 insertions(+), 17 deletions(-) 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 115e53171067..56da71801e17 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 @@ -551,8 +551,8 @@ public SnapshotDiffResponse getSnapshotDiffReport( snapDiffJob.getReason()); case DONE: SnapshotDiffReportOzone report = createPageResponse(snapDiffJob, - volumeName, bucketName, fromSnapshotName, toSnapshotName, index, - pageSize); + snapDiffJobKey, volumeName, bucketName, fromSnapshotName, + toSnapshotName, index, pageSize); return new SnapshotDiffResponse(report, DONE, 0L); case REJECTED: return new SnapshotDiffResponse( @@ -624,8 +624,8 @@ public SnapshotDiffResponse getSnapshotDiffReport( snapDiffJob.getReason()); case DONE: SnapshotDiffReportOzone report = createPageResponse(snapDiffJob, - volumeName, bucketName, fromSnapshotName, toSnapshotName, index, - pageSize); + snapDiffJobKey, volumeName, bucketName, fromSnapshotName, + toSnapshotName, index, pageSize); return new SnapshotDiffResponse(report, DONE, 0L); case CANCELLED: return new SnapshotDiffResponse( @@ -708,8 +708,10 @@ public OFSPath getSnapshotRootPath(String volume, String bucket) { } @VisibleForTesting + @SuppressWarnings("parameternumber") SnapshotDiffReportOzone createPageResponse( final SnapshotDiffJob snapDiffJob, + final String jobKey, final String volumeName, final String bucketName, final String fromSnapshotName, @@ -748,7 +750,7 @@ SnapshotDiffReportOzone createPageResponse( OFSPath path = getSnapshotRootPath(volumeName, bucketName); Pair, String> pageResponse = - createPageResponse(snapDiffJob, index, pageSize); + createPageResponse(snapDiffJob, jobKey, index, pageSize); List diffReportList = pageResponse.getLeft(); String tokenString = pageResponse.getRight(); @@ -758,6 +760,7 @@ SnapshotDiffReportOzone createPageResponse( Pair, String> createPageResponse( final SnapshotDiffJob snapDiffJob, + final String jobKey, final String index, final int pageSize ) throws IOException { @@ -784,7 +787,7 @@ Pair, String> createPageResponse( String nextTokenString = iterator.hasNext() ? getIndexFromReportKey(StringCodec.get().fromPersistedFormat(iterator.next().getKey())) : null; - checkReportsIntegrity(snapDiffJob, pageLastKey, nextTokenString == null); + checkReportsIntegrity(snapDiffJob, jobKey, pageLastKey, nextTokenString == null); return Pair.of(diffReportList, nextTokenString); } @@ -799,13 +802,19 @@ Pair, String> createPageResponse( */ @VisibleForTesting void checkReportsIntegrity(final SnapshotDiffJob diffJob, + final String jobKey, final String largestPageIndex, boolean lastPage) throws IOException { - // For last page check last entry returned if the largest entry key equals the largest key stored in the job entry. - if (lastPage && (diffJob.getLargestEntryKey() == null || !diffJob.getLargestEntryKey().equals(largestPageIndex))) { + // A DONE job persisted before largestEntryKey was introduced deserializes with a + // null largestEntryKey. Such a legacy job carries no stored last-entry to compare, + // so skip the integrity assertion for it rather than treating it as a failure. + if (lastPage && diffJob.getLargestEntryKey() != null && + !diffJob.getLargestEntryKey().equals(largestPageIndex)) { LOG.error("Expected last entry: {} but found " + "Largest Page entry: {}", diffJob.getLargestEntryKey(), largestPageIndex); - updateJobStatus(diffJob.getJobId(), DONE, FAILED); + // The job is stored under jobKey (fromSnapshotId + DELIMITER + toSnapshotId), not + // under its jobId; passing the jobId here would look up a missing row. + updateJobStatus(jobKey, DONE, FAILED); throw new IOException("Report integrity check failed. Retry after: " + ozoneManager.getOmSnapshotManager().getDiffCleanupServiceInterval()); } @@ -846,8 +855,8 @@ private synchronized SnapshotDiffResponse submitSnapDiffJob( if (snapDiffJob.getStatus() != QUEUED) { // Same request is submitted by another thread and already completed. if (snapDiffJob.getStatus() == DONE) { - SnapshotDiffReportOzone report = createPageResponse(snapDiffJob, volume, - bucket, fromSnapshot, toSnapshot, index, pageSize); + SnapshotDiffReportOzone report = createPageResponse(snapDiffJob, jobKey, + volume, bucket, fromSnapshot, toSnapshot, index, pageSize); return new SnapshotDiffResponse(report, DONE, 0L); } else { // Otherwise, return the same status as in DB with wait time. @@ -1718,6 +1727,11 @@ private synchronized void updateJobStatus(String jobKey, JobStatus oldStatus, JobStatus newStatus) { SnapshotDiffJob snapshotDiffJob = snapDiffJobTable.get(jobKey); + if (snapshotDiffJob == null) { + LOG.warn("Cannot transition snapshot diff job {} from {} to {}: " + + "no job found for the key.", jobKey, oldStatus, newStatus); + return; + } if (snapshotDiffJob.getStatus() != oldStatus) { throw new IllegalStateException("Invalid job status for jobID: " + snapshotDiffJob.getJobId() + ". Job's current status is '" + 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 3981f8bf1a05..1318f59ae7fc 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 @@ -659,7 +659,7 @@ public void testGenerateDiffReport() throws IOException { JobStatus.DONE, "vol", "buck", "fs", "ts", false, true, diffMap.size(), null, 0.0, totalDiffEntries.getRight()); SnapshotDiffReportOzone snapshotDiffReportOzone = - snapshotDiffManager.createPageResponse(snapshotDiffJob, "vol", + snapshotDiffManager.createPageResponse(snapshotDiffJob, "jobId", "vol", "buck", "fs", "ts", "", Integer.MAX_VALUE); Set expectedOrder = new LinkedHashSet<>(); @@ -751,12 +751,12 @@ public void testCreatePageResponse(int startIdx, if (pageSize <= 0) { int finalPageSize = pageSize; assertThrows(IOException.class, - () -> snapshotDiffManager.createPageResponse(snapshotDiffJob, "vol", + () -> snapshotDiffManager.createPageResponse(snapshotDiffJob, testJobId, "vol", "buck", "fs", "ts", pageMap.get(startIdx), finalPageSize)); continue; } SnapshotDiffReportOzone snapshotDiffReportOzone = - snapshotDiffManager.createPageResponse(snapshotDiffJob, "vol", + snapshotDiffManager.createPageResponse(snapshotDiffJob, testJobId, "vol", "buck", "fs", "ts", pageMap.get(startIdx), pageSize); int expectedTotalNumberOfRecords = @@ -774,6 +774,87 @@ public void testCreatePageResponse(int startIdx, } } + /** + * A DONE job persisted before the largestEntryKey field existed deserializes + * with a null largestEntryKey after an upgrade. Reading its (fully intact) + * report must not trip the integrity check: the job stays readable instead of + * hitting an NPE in the recovery path. + */ + @Test + public void testCreatePageResponseLegacyNullLargestEntryKeyStaysReadable() + throws IOException, RocksDBException { + UUID fromSnapshotId = UUID.randomUUID(); + UUID toSnapshotId = UUID.randomUUID(); + String jobKey = fromSnapshotId + DELIMITER + toSnapshotId; + String jobId = UUID.randomUUID().toString(); + assertThat(jobId).isNotEqualTo(jobKey); + + DiffReportEntry reportEntry = new DiffReportEntry( + SnapshotDiffReport.DiffType.MODIFY, codecRegistry.asRawData("key0")); + String reportKey = SnapshotDiffManager.getReportKeyForIndex( + jobId, SnapshotDiffReport.DiffType.MODIFY, 0); + db.get().put(snapDiffReportTable, codecRegistry.asRawData(reportKey), + codecRegistry.asRawData(reportEntry)); + + // Round-trip through the production codec to confirm a legacy job deserializes + // with a null largestEntryKey (this is byte-for-byte what an upgraded OM loads). + SnapshotDiffJob legacyJob = new SnapshotDiffJob(0, jobId, DONE, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", false, true, 1, null, 0.0, null); + SnapshotDiffJob loaded = SnapshotDiffJob.codec().fromPersistedFormat( + SnapshotDiffJob.codec().toPersistedFormat(legacyJob)); + assertNull(loaded.getLargestEntryKey()); + db.get().put(snapDiffJobTable, codecRegistry.asRawData(jobKey), + codecRegistry.asRawData(loaded)); + + SnapshotDiffReportOzone report = snapshotDiffManager.createPageResponse(loaded, + jobKey, VOLUME_NAME, BUCKET_NAME, "fs", "ts", "", Integer.MAX_VALUE); + assertThat(report.getDiffList()).hasSize(1); + SnapshotDiffJob afterJob = codecRegistry.asObject( + db.get().get(snapDiffJobTable, codecRegistry.asRawData(jobKey)), + SnapshotDiffJob.class); + assertEquals(DONE, afterJob.getStatus()); + } + + /** + * When a DONE job's stored largestEntryKey genuinely disagrees with the report + * tail, the integrity check must flip the job (looked up by its dedup jobKey, + * not by the per-generation jobId) to FAILED and throw the designed retriable + * IOException rather than NPE on a wrong-key lookup. + */ + @Test + public void testCreatePageResponseIntegrityFailureUsesJobKey() + throws IOException, RocksDBException { + UUID fromSnapshotId = UUID.randomUUID(); + UUID toSnapshotId = UUID.randomUUID(); + String jobKey = fromSnapshotId + DELIMITER + toSnapshotId; + String jobId = UUID.randomUUID().toString(); + assertThat(jobId).isNotEqualTo(jobKey); + + DiffReportEntry reportEntry = new DiffReportEntry( + SnapshotDiffReport.DiffType.MODIFY, codecRegistry.asRawData("key0")); + String reportKey = SnapshotDiffManager.getReportKeyForIndex( + jobId, SnapshotDiffReport.DiffType.MODIFY, 0); + db.get().put(snapDiffReportTable, codecRegistry.asRawData(reportKey), + codecRegistry.asRawData(reportEntry)); + + // Non-null largestEntryKey that does not match the actual report tail. + SnapshotDiffJob doneJob = new SnapshotDiffJob(0, jobId, DONE, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", false, true, 1, null, 0.0, + jobId + DELIMITER + "mismatched-tail"); + db.get().put(snapDiffJobTable, codecRegistry.asRawData(jobKey), + codecRegistry.asRawData(doneJob)); + + IOException ex = assertThrows(IOException.class, () -> + snapshotDiffManager.createPageResponse(doneJob, jobKey, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", "", Integer.MAX_VALUE)); + assertThat(ex).hasMessageContaining("Report integrity check failed"); + + SnapshotDiffJob afterJob = codecRegistry.asObject( + db.get().get(snapDiffJobTable, codecRegistry.asRawData(jobKey)), + SnapshotDiffJob.class); + assertEquals(FAILED, afterJob.getStatus()); + } + @Test public void testCreatePageResponseDependencyOrderedReportKeys() throws Exception { String testJobId = "dependencyJob"; @@ -794,14 +875,14 @@ public void testCreatePageResponseDependencyOrderedReportKeys() throws Exception testJobId + DELIMITER + largestIndexToken); SnapshotDiffReportOzone firstPage = snapshotDiffManager.createPageResponse(snapshotDiffJob, - "vol", "buck", "fs", "ts", "", 2); + testJobId, "vol", "buck", "fs", "ts", "", 2); assertEquals(2, firstPage.getDiffList().size()); assertEquals(getTestDiffEntry(testJobId, 0), firstPage.getDiffList().get(0)); assertEquals(getTestDiffEntry(testJobId, 1), firstPage.getDiffList().get(1)); assertEquals(orderedIndexTokens.get(2), firstPage.getToken()); SnapshotDiffReportOzone secondPage = snapshotDiffManager.createPageResponse(snapshotDiffJob, - "vol", "buck", "fs", "ts", firstPage.getToken(), 2); + testJobId, "vol", "buck", "fs", "ts", firstPage.getToken(), 2); assertEquals(1, secondPage.getDiffList().size()); assertEquals(getTestDiffEntry(testJobId, 2), secondPage.getDiffList().get(0)); assertNull(secondPage.getToken()); @@ -1761,7 +1842,7 @@ public void testGetSnapshotDiffReportWhenDone() throws Exception { ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName, ctx.toSnapshotName, expectedEntries, null); doReturn(dummyReport).when(spy).createPageResponse(any(SnapshotDiffJob.class), - eq(ctx.volumeName), eq(ctx.bucketName), eq(ctx.fromSnapshotName), + eq(ctx.diffJobKey), eq(ctx.volumeName), eq(ctx.bucketName), eq(ctx.fromSnapshotName), eq(ctx.toSnapshotName), eq(""), eq(1000)); SnapshotDiffJob doneJob = new SnapshotDiffJob(0L, jobId, DONE, ctx.volumeName, From 11669b291f67283380e531a65b5c578415d583c6 Mon Sep 17 00:00:00 2001 From: Wei-Chiu Chuang Date: Tue, 27 Oct 2026 17:20:20 -0700 Subject: [PATCH 2/3] HDDS-16442. Guard integrity failures against replacement jobs Change-Id: I9a067647517f2ae511f6d5aa191d6ae552340593 --- .../om/snapshot/SnapshotDiffManager.java | 12 +++--- .../om/snapshot/TestSnapshotDiffManager.java | 42 +++++++++++++++++-- 2 files changed, 44 insertions(+), 10 deletions(-) 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 56da71801e17..00cff768d6d4 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 @@ -801,7 +801,7 @@ Pair, String> createPageResponse( * service and throws the exception to client. */ @VisibleForTesting - void checkReportsIntegrity(final SnapshotDiffJob diffJob, + synchronized void checkReportsIntegrity(final SnapshotDiffJob diffJob, final String jobKey, final String largestPageIndex, boolean lastPage) throws IOException { @@ -814,7 +814,10 @@ void checkReportsIntegrity(final SnapshotDiffJob diffJob, "Largest Page entry: {}", diffJob.getLargestEntryKey(), largestPageIndex); // The job is stored under jobKey (fromSnapshotId + DELIMITER + toSnapshotId), not // under its jobId; passing the jobId here would look up a missing row. - updateJobStatus(jobKey, DONE, FAILED); + SnapshotDiffJob currentJob = snapDiffJobTable.get(jobKey); + if (currentJob != null && currentJob.getJobId().equals(diffJob.getJobId()) && currentJob.getStatus() == DONE) { + updateJobStatus(jobKey, DONE, FAILED); + } throw new IOException("Report integrity check failed. Retry after: " + ozoneManager.getOmSnapshotManager().getDiffCleanupServiceInterval()); } @@ -1727,11 +1730,6 @@ private synchronized void updateJobStatus(String jobKey, JobStatus oldStatus, JobStatus newStatus) { SnapshotDiffJob snapshotDiffJob = snapDiffJobTable.get(jobKey); - if (snapshotDiffJob == null) { - LOG.warn("Cannot transition snapshot diff job {} from {} to {}: " + - "no job found for the key.", jobKey, oldStatus, newStatus); - return; - } if (snapshotDiffJob.getStatus() != oldStatus) { throw new IllegalStateException("Invalid job status for jobID: " + snapshotDiffJob.getJobId() + ". Job's current status is '" + 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 1318f59ae7fc..f8fdf6daf533 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 @@ -796,8 +796,7 @@ public void testCreatePageResponseLegacyNullLargestEntryKeyStaysReadable() db.get().put(snapDiffReportTable, codecRegistry.asRawData(reportKey), codecRegistry.asRawData(reportEntry)); - // Round-trip through the production codec to confirm a legacy job deserializes - // with a null largestEntryKey (this is byte-for-byte what an upgraded OM loads). + // Round-trip a completed job without a stored last-entry key through the production codec. SnapshotDiffJob legacyJob = new SnapshotDiffJob(0, jobId, DONE, VOLUME_NAME, BUCKET_NAME, "fs", "ts", false, true, 1, null, 0.0, null); SnapshotDiffJob loaded = SnapshotDiffJob.codec().fromPersistedFormat( @@ -808,7 +807,7 @@ public void testCreatePageResponseLegacyNullLargestEntryKeyStaysReadable() SnapshotDiffReportOzone report = snapshotDiffManager.createPageResponse(loaded, jobKey, VOLUME_NAME, BUCKET_NAME, "fs", "ts", "", Integer.MAX_VALUE); - assertThat(report.getDiffList()).hasSize(1); + assertThat(report.getDiffList()).containsExactly(reportEntry); SnapshotDiffJob afterJob = codecRegistry.asObject( db.get().get(snapDiffJobTable, codecRegistry.asRawData(jobKey)), SnapshotDiffJob.class); @@ -855,6 +854,43 @@ public void testCreatePageResponseIntegrityFailureUsesJobKey() assertEquals(FAILED, afterJob.getStatus()); } + @ParameterizedTest + @CsvSource({"DONE,true", "IN_PROGRESS,true", "QUEUED,true", "FAILED,true", "FAILED,false"}) + public void testIntegrityFailureDoesNotUpdateChangedJob(JobStatus currentStatus, boolean replacement) + throws IOException { + String jobKey = UUID.randomUUID() + DELIMITER + UUID.randomUUID(); + String jobId = UUID.randomUUID().toString(); + SnapshotDiffJob oldJob = new SnapshotDiffJob(0, jobId, DONE, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", false, true, 1, null, 0.0, jobId + DELIMITER + "missing-tail"); + String currentJobId = replacement ? UUID.randomUUID().toString() : jobId; + SnapshotDiffJob currentJob = new SnapshotDiffJob(0, currentJobId, currentStatus, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", false, true, 0, null, 0.0, ""); + snapshotDiffManager.getSnapDiffJobTable().put(jobKey, currentJob); + + IOException ex = assertThrows(IOException.class, () -> + snapshotDiffManager.createPageResponse(oldJob, jobKey, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", "", Integer.MAX_VALUE)); + assertThat(ex).hasMessageContaining("Report integrity check failed"); + SnapshotDiffJob afterJob = snapshotDiffManager.getSnapDiffJobTable().get(jobKey); + assertEquals(currentJobId, afterJob.getJobId()); + assertEquals(currentStatus, afterJob.getStatus()); + assertEquals("", afterJob.getLargestEntryKey()); + } + + @Test + public void testIntegrityFailureDoesNotRecreateRemovedJob() throws IOException { + String jobKey = UUID.randomUUID() + DELIMITER + UUID.randomUUID(); + String jobId = UUID.randomUUID().toString(); + SnapshotDiffJob oldJob = new SnapshotDiffJob(0, jobId, DONE, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", false, true, 1, null, 0.0, jobId + DELIMITER + "missing-tail"); + + IOException ex = assertThrows(IOException.class, () -> + snapshotDiffManager.createPageResponse(oldJob, jobKey, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", "", Integer.MAX_VALUE)); + assertThat(ex).hasMessageContaining("Report integrity check failed"); + assertNull(snapshotDiffManager.getSnapDiffJobTable().get(jobKey)); + } + @Test public void testCreatePageResponseDependencyOrderedReportKeys() throws Exception { String testJobId = "dependencyJob"; From 222cb2b0c507eb61a7bdb4e375159b2fd237dbae Mon Sep 17 00:00:00 2001 From: Wei-Chiu Chuang Date: Tue, 27 Oct 2026 19:52:44 -0700 Subject: [PATCH 3/3] HDDS-16442. Serialize integrity failure recovery with cleanup Change-Id: If27454033da53cc3134539bdef0d145aa00ff707 --- .../service/SnapshotDiffCleanupService.java | 6 ++- .../om/snapshot/SnapshotDiffManager.java | 9 ++-- .../om/snapshot/TestSnapshotDiffManager.java | 51 ++++++++++++++++++- 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDiffCleanupService.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDiffCleanupService.java index 9aaeafa1c962..b3a7658d5a69 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDiffCleanupService.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDiffCleanupService.java @@ -124,7 +124,11 @@ public void run() { // assumption is that by the next cleanup run, there is no purged snapDiff // job reading from report table. long purgedReportJobs = removeOlderJobReport(); - long movedJobsToPurgeTable = moveOldSnapDiffJobsToPurgeTable(); + long movedJobsToPurgeTable; + // Serialize job deletion with SnapshotDiffManager's integrity-failure status transition. + synchronized (db) { + movedJobsToPurgeTable = moveOldSnapDiffJobsToPurgeTable(); + } LOG.info("Snapshot diff cleanup run completed. Purged report jobs: {}, " + "moved jobs to purge table: {}.", purgedReportJobs, movedJobsToPurgeTable); 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 00cff768d6d4..1c2491671a68 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 @@ -814,9 +814,12 @@ synchronized void checkReportsIntegrity(final SnapshotDiffJob diffJob, "Largest Page entry: {}", diffJob.getLargestEntryKey(), largestPageIndex); // The job is stored under jobKey (fromSnapshotId + DELIMITER + toSnapshotId), not // under its jobId; passing the jobId here would look up a missing row. - SnapshotDiffJob currentJob = snapDiffJobTable.get(jobKey); - if (currentJob != null && currentJob.getJobId().equals(diffJob.getJobId()) && currentJob.getStatus() == DONE) { - updateJobStatus(jobKey, DONE, FAILED); + // Cleanup deletes job rows without the manager lock, so share the DB monitor with cleanup as well. + synchronized (db) { + SnapshotDiffJob currentJob = snapDiffJobTable.get(jobKey); + if (currentJob != null && currentJob.getJobId().equals(diffJob.getJobId()) && currentJob.getStatus() == DONE) { + updateJobStatus(jobKey, DONE, FAILED); + } } throw new IOException("Report integrity check failed. Retry after: " + ozoneManager.getOmSnapshotManager().getDiffCleanupServiceInterval()); 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 f8fdf6daf533..ba0516b853f3 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 @@ -96,6 +96,7 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; @@ -135,6 +136,7 @@ import org.apache.hadoop.ozone.om.helpers.SnapshotInfo; import org.apache.hadoop.ozone.om.helpers.WithParentObjectId; import org.apache.hadoop.ozone.om.lock.OmReadOnlyLock; +import org.apache.hadoop.ozone.om.service.SnapshotDiffCleanupService; import org.apache.hadoop.ozone.om.snapshot.SnapshotTestUtils.StubbedPersistentMap; import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse; import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse.CancelMessage; @@ -159,6 +161,7 @@ import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; import org.mockito.Mock; import org.mockito.MockedConstruction; import org.mockito.MockedStatic; @@ -183,6 +186,7 @@ public class TestSnapshotDiffManager { private List columnFamilyHandles; private ColumnFamilyHandle snapDiffJobTable; private ColumnFamilyHandle snapDiffReportTable; + private ColumnFamilyHandle snapDiffPurgedJobTable; private SnapshotDiffManager snapshotDiffManager; private final List jobStatuses = Arrays.asList(QUEUED, IN_PROGRESS, DONE, REJECTED, FAILED); @@ -258,7 +262,7 @@ public void init() throws RocksDBException, IOException, ExecutionException { new ColumnFamilyDescriptor( StringUtils.string2Bytes(SNAP_DIFF_REPORT_TABLE_NAME), columnFamilyOptions)); - ColumnFamilyHandle snapDiffPurgedJobTable = db.get().createColumnFamily( + snapDiffPurgedJobTable = db.get().createColumnFamily( new ColumnFamilyDescriptor( StringUtils.string2Bytes(SNAP_DIFF_PURGED_JOB_TABLE_NAME), columnFamilyOptions)); @@ -891,6 +895,51 @@ public void testIntegrityFailureDoesNotRecreateRemovedJob() throws IOException { assertNull(snapshotDiffManager.getSnapDiffJobTable().get(jobKey)); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testIntegrityFailureAndCleanupUseSameLock(boolean cleanupFirst) throws Exception { + String jobKey = UUID.randomUUID() + DELIMITER + UUID.randomUUID(); + String jobId = UUID.randomUUID().toString(); + SnapshotDiffJob oldJob = new SnapshotDiffJob(0, jobId, DONE, VOLUME_NAME, + BUCKET_NAME, "fs", "ts", false, true, 1, null, 0.0, jobId + DELIMITER + "missing-tail"); + snapshotDiffManager.getSnapDiffJobTable().put(jobKey, oldJob); + when(ozoneManager.getConfiguration()).thenReturn(new OzoneConfiguration()); + SnapshotDiffCleanupService cleanup = new SnapshotDiffCleanupService(1000, 1000, ozoneManager, db, + snapDiffJobTable, snapDiffPurgedJobTable, snapDiffReportTable, codecRegistry); + FutureTask task = new FutureTask<>(() -> { + if (cleanupFirst) { + cleanup.run(); + } else { + assertThrows(IOException.class, () -> snapshotDiffManager.checkReportsIntegrity(oldJob, jobKey, "", true)); + } + return null; + }); + Thread worker = new Thread(task); + try { + synchronized (db) { + worker.start(); + attempt(() -> { + if (worker.getState() != Thread.State.BLOCKED) { + throw new IllegalStateException("Worker has not reached the DB monitor yet."); + } + return null; + }, 100, TimeDuration.valueOf(10, TimeUnit.MILLISECONDS), null, null); + if (cleanupFirst) { + assertThrows(IOException.class, () -> snapshotDiffManager.checkReportsIntegrity(oldJob, jobKey, "", true)); + assertEquals(FAILED, snapshotDiffManager.getSnapDiffJobTable().get(jobKey).getStatus()); + } else { + cleanup.run(); + assertNull(snapshotDiffManager.getSnapDiffJobTable().get(jobKey)); + } + } + task.get(10, TimeUnit.SECONDS); + assertNull(snapshotDiffManager.getSnapDiffJobTable().get(jobKey)); + } finally { + worker.join(10000); + cleanup.shutdown(); + } + } + @Test public void testCreatePageResponseDependencyOrderedReportKeys() throws Exception { String testJobId = "dependencyJob";