Skip to content

HDDS-16442. Fix snapshot diff integrity failure recovery after upgrade #11808

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jojochuang wants to merge 3 commits into
base: master
Choose a base branch
from
Draft

HDDS-16442. Fix snapshot diff integrity failure recovery after upgrade #11808

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e003eeb
HDDS-16442. Fix snapshot diff integrity failure recovery after upgrade
jojochuang Oct 27, 2026
11669b2
HDDS-16442. Guard integrity failures against replacement jobs
jojochuang Oct 28, 2026
222cb2b
HDDS-16442. Serialize integrity failure recovery with cleanup
jojochuang Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...-manager/src/main/java/org/apache/hadoop/ozone/om/service/SnapshotDiffCleanupService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
39 changes: 27 additions & 12 deletions .../ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -748,7 +750,7 @@ SnapshotDiffReportOzone createPageResponse(
OFSPath path = getSnapshotRootPath(volumeName, bucketName);

Pair<List<DiffReportEntry>, String> pageResponse =
createPageResponse(snapDiffJob, index, pageSize);
createPageResponse(snapDiffJob, jobKey, index, pageSize);
List<DiffReportEntry> diffReportList = pageResponse.getLeft();
String tokenString = pageResponse.getRight();

Expand All @@ -758,6 +760,7 @@ SnapshotDiffReportOzone createPageResponse(

Pair<List<DiffReportEntry>, String> createPageResponse(
final SnapshotDiffJob snapDiffJob,
final String jobKey,
final String index,
final int pageSize
) throws IOException {
Expand All @@ -784,7 +787,7 @@ Pair<List<DiffReportEntry>, 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);
}

Expand All @@ -798,14 +801,26 @@ Pair<List<DiffReportEntry>, 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 {
// 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.
// 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());
}
Expand Down Expand Up @@ -846,8 +861,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.
Expand Down
180 changes: 173 additions & 7 deletions ...ne-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -183,6 +186,7 @@ public class TestSnapshotDiffManager {
private List<ColumnFamilyHandle> columnFamilyHandles;
private ColumnFamilyHandle snapDiffJobTable;
private ColumnFamilyHandle snapDiffReportTable;
private ColumnFamilyHandle snapDiffPurgedJobTable;
private SnapshotDiffManager snapshotDiffManager;
private final List<JobStatus> jobStatuses = Arrays.asList(QUEUED, IN_PROGRESS,
DONE, REJECTED, FAILED);
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -659,7 +663,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<SnapshotDiffReport.DiffType> expectedOrder = new LinkedHashSet<>();
Expand Down Expand Up @@ -751,12 +755,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 =
Expand All @@ -774,6 +778,168 @@ 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 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(
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()).containsExactly(reportEntry);
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());
}

@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));
}

@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<Void> 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";
Expand All @@ -794,14 +960,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());
Expand Down Expand Up @@ -1761,7 +1927,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,
Expand Down
Loading