Skip to content

HDDS-16089. Improve MPU table evaluation efficiency in KeyLifeCycleService #10949

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

Merged
ChenSammi merged 3 commits into from
Aug 14, 2026
+190 −62
Merged

HDDS-16089. Improve MPU table evaluation efficiency in KeyLifeCycleService #10949

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
30aef6b
HDDS-16089. Improve MPU table evaluation efficiency in LifeCycleService
priyeshkaratha Aug 4, 2026
015f57a
addressing review comments
priyeshkaratha Aug 10, 2026
fadfc90
addressing review comments
priyeshkaratha Aug 13, 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
133 changes: 71 additions & 62 deletions ...e/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1109,55 +1109,81 @@ private void processMultipartUploads(OmBucketInfo bucketInfo, List<OmLCRule> rul
upload.setCreationTime(Instant.ofEpochMilli(mpuKeyInfo.getCreationTime()));
String keyName = upload.getKeyName();

String multipartOpenKey;
try {
multipartOpenKey = OMMultipartUploadUtils.getMultipartOpenKey(
volumeName, bucketName, keyName, upload.getUploadId(),
omMetadataManager, bucketInfo.getBucketLayout());
} catch (OMException e) {
LOG.warn("Failed to get multipart open key for {}/{}/{}, skipping",
volumeName, bucketName, keyName, e);
continue;
}

OmKeyInfo openKeyInfo = omMetadataManager.getOpenKeyTable(bucketInfo.getBucketLayout())
.get(multipartOpenKey);
if (openKeyInfo == null) {
LOG.warn("Open key not found for multipart upload {}/{}/{}, skipping",
volumeName, bucketName, keyName);
continue;
}

OmLCRule matchingRule = null;
OmKeyInfo openKeyInfo = null;
boolean openKeyFetchAttempted = false;
boolean skipUpload = false;
for (OmLCRule rule : ruleList) {
if (shouldAbortUpload(openKeyInfo, upload, keyName, rule)) {
if (expiredUploads.isFull()) {
LOG.info("Multipart upload batch reached part count limit {}, aborting current batch " +
"({} uploads, {} parts) for bucket {}/{}",
mpuAbortLimitPerTask, expiredUploads.size(), expiredUploads.getPartCount(),
volumeName, bucketName);
abortExpiredMultipartUploadsAndClear(bucketInfo, expiredUploads);
}

// Split-schema MPUs keep parts in multipartPartsTable (the embedded map
// is empty); legacy MPUs use the embedded map. An MPU with no uploaded
// parts is valid (S3 allows aborting it with an empty parts list).
int uploadedParts;
if (!passesAgeAndPrefix(upload, keyName, rule)) {
continue;
}
if (!rule.isTagEnable()) {
Comment thread
priyeshkaratha marked this conversation as resolved.
matchingRule = rule;
break;
}
if (!openKeyFetchAttempted) {
openKeyFetchAttempted = true;
try {
uploadedParts = mpuKeyInfo.getSchemaVersion()
== OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION
? OMMultipartUploadUtils.countParts(omMetadataManager, upload.getUploadId())
: mpuKeyInfo.getPartKeyInfoMap().size();
String multipartOpenKey = OMMultipartUploadUtils.getMultipartOpenKey(
volumeName, bucketName, keyName, upload.getUploadId(),
omMetadataManager, bucketInfo.getBucketLayout());
openKeyInfo = omMetadataManager.getOpenKeyTable(
bucketInfo.getBucketLayout()).get(multipartOpenKey);
} catch (OMException e) {
LOG.warn("Failed to get multipart open key for {}/{}/{}, skipping",
volumeName, bucketName, keyName, e);
skipUpload = true;
break;
} catch (IOException e) {
LOG.warn("Failed to count parts for MPU {}/{}/{} uploadId {}, skipping",
volumeName, bucketName, keyName, upload.getUploadId(), e);
LOG.warn("Failed to read open key table for {}/{}/{}, skipping",
volumeName, bucketName, keyName, e);
skipUpload = true;
break;
}
expiredUploads.add(upload, uploadedParts);
LOG.debug("Multipart upload {}/{}/{} with uploadId {} ({} parts) will be aborted",
volumeName, bucketName, keyName, upload.getUploadId(), uploadedParts);
if (openKeyInfo == null) {
LOG.debug("Orphan multipart upload {}/{}/{} has no open key entry, skipping tag-requiring rules",
volumeName, bucketName, keyName);
}
}
if (openKeyInfo == null) {
continue;
}
OmLCFilter filter = rule.getFilter();
if (filter == null || filter.match(openKeyInfo, keyName)) {
matchingRule = rule;
break;
}
}

if (skipUpload || matchingRule == null) {
continue;
}

if (expiredUploads.isFull()) {
LOG.info("Multipart upload batch reached part count limit {}, aborting current batch " +
"({} uploads, {} parts) for bucket {}/{}",
mpuAbortLimitPerTask, expiredUploads.size(), expiredUploads.getPartCount(),
volumeName, bucketName);
abortExpiredMultipartUploadsAndClear(bucketInfo, expiredUploads);
}

// Split-schema MPUs keep parts in multipartPartsTable (the embedded map
// is empty); legacy MPUs use the embedded map. An MPU with no uploaded
// parts is valid (S3 allows aborting it with an empty parts list).
int uploadedParts;
try {
uploadedParts = mpuKeyInfo.getSchemaVersion()
== OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION
? OMMultipartUploadUtils.countParts(omMetadataManager, upload.getUploadId())
: mpuKeyInfo.getPartKeyInfoMap().size();
} catch (IOException e) {
LOG.warn("Failed to count parts for MPU {}/{}/{} uploadId {}, skipping",
volumeName, bucketName, keyName, upload.getUploadId(), e);
continue;
}
expiredUploads.add(upload, uploadedParts);
LOG.debug("Multipart upload {}/{}/{} with uploadId {} ({} parts) will be aborted",
volumeName, bucketName, keyName, upload.getUploadId(), uploadedParts);
}
} catch (IOException e) {
LOG.warn("Failed to iterate multipartInfoTable for bucket {}/{}", volumeName, bucketName, e);
Expand All @@ -1172,33 +1198,16 @@ private void processMultipartUploads(OmBucketInfo bucketInfo, List<OmLCRule> rul
}

/**
* Check if a multipart upload should be aborted based on the lifecycle rule.
*
* @param openKeyInfo the open key information with tags
* @param upload the multipart upload information
* @param keyName the key name of the upload
* @param rule the lifecycle rule to evaluate against
* @return true if the upload should be aborted, false otherwise
* Returns true if the upload passes the age and prefix checks for the given rule,
* without consulting the open key table (no tag evaluation).
*/
private boolean shouldAbortUpload(OmKeyInfo openKeyInfo, OmMultipartUpload upload,
String keyName, OmLCRule rule) {

private boolean passesAgeAndPrefix(OmMultipartUpload upload, String keyName, OmLCRule rule) {
if (!rule.getAbortIncompleteMultipartUpload().shouldAbort(
upload.getCreationTime().toEpochMilli())) {
return false;
}

String effectivePrefix = rule.getEffectivePrefix();
if (effectivePrefix != null && !keyName.startsWith(effectivePrefix)) {
return false;
}

OmLCFilter filter = rule.getFilter();
if (filter != null && !filter.match(openKeyInfo, keyName)) {
return false;
}

return true;
return effectivePrefix == null || keyName.startsWith(effectivePrefix);
}

/**
Expand Down
119 changes: 119 additions & 0 deletions ...one-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2834,6 +2834,125 @@ void testAbortIncompleteMultipartUploadWithTagFilter() throws Exception {
deleteLifecyclePolicy(volumeName, bucketName);
}

/**
* An MPU whose open key entry is missing (orphan) should still be aborted when a lifecycle
* rule matches by age and prefix alone (no tag filter). The abort request handler already
* tolerates a missing open key.
*/
@Test
void testOrphanMpuAbortedByAgeAndPrefixRule() throws Exception {
final String volumeName = getTestName();
final String bucketName = uniqueObjectName("bucket");

createVolumeAndBucket(volumeName, bucketName, OBJECT_STORE,
UserGroupInformation.getCurrentUser().getShortUserName());

String owner = UserGroupInformation.getCurrentUser().getShortUserName();
long initialMpuCount = getMultipartUploadCount(volumeName, bucketName);

// Create two MPUs: one normal (has open key), one will become an orphan.
OmMultipartInfo normalMpu = createTestMultipartUpload(volumeName, bucketName, "data/normal", owner);
OmMultipartInfo orphanMpu = createTestMultipartUpload(volumeName, bucketName, "data/orphan", owner);

// Age both MPUs past the threshold.
long oldCreationTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
updateMultipartUploadCreationTime(volumeName, bucketName, "data/normal",
normalMpu.getUploadID(), oldCreationTime);
updateMultipartUploadCreationTime(volumeName, bucketName, "data/orphan",
orphanMpu.getUploadID(), oldCreationTime);

// Simulate orphan: delete the open key entry for "data/orphan" directly from the table.
// Use the same helper the service uses so the key format matches exactly.
String resolvedOrphanOpenKey = OMMultipartUploadUtils
.getMultipartOpenKey(volumeName, bucketName, "data/orphan", orphanMpu.getUploadID(),
metadataManager, OBJECT_STORE);
metadataManager.getOpenKeyTable(OBJECT_STORE).delete(resolvedOrphanOpenKey);

// Rule: abort all MPUs under "data/" after 1 day — no tag filter.
OmLCRule rule = new OmLCRule.Builder()
.setId("abort-data-prefix")
.setEnabled(true)
.setFilter(new OmLCFilter.Builder().setPrefix("data/").build())
.setAction(new OmLCAbortIncompleteMultipartUpload.Builder()
.setDaysAfterInitiation(1)
.build())
.build();

createLifecyclePolicy(volumeName, bucketName, OBJECT_STORE, Collections.singletonList(rule));

// Both MPUs should be aborted: normal one (open key present) and orphan (open key missing).
GenericTestUtils.waitFor(() ->
getMultipartUploadCount(volumeName, bucketName) - initialMpuCount == 0,
WAIT_CHECK_INTERVAL, 10000);

String normalKey = metadataManager.getMultipartKey(volumeName, bucketName,
"data/normal", normalMpu.getUploadID());
assertNull(metadataManager.getMultipartInfoTable().get(normalKey),
"Normal MPU should be aborted");

String orphanKey = metadataManager.getMultipartKey(volumeName, bucketName,
"data/orphan", orphanMpu.getUploadID());
assertNull(metadataManager.getMultipartInfoTable().get(orphanKey),
"Orphan MPU (no open key) should be aborted when a tag-free rule matches");

deleteLifecyclePolicy(volumeName, bucketName);
}

/**
* An MPU whose open key is missing (orphan) must NOT be aborted when the only matching
* lifecycle rule requires a tag filter, because the tag metadata is unavailable.
*/
@Test
void testOrphanMpuNotAbortedByTagOnlyRule() throws Exception {
final String volumeName = getTestName();
final String bucketName = uniqueObjectName("bucket");

createVolumeAndBucket(volumeName, bucketName, OBJECT_STORE,
UserGroupInformation.getCurrentUser().getShortUserName());

String owner = UserGroupInformation.getCurrentUser().getShortUserName();
long initialMpuCount = getMultipartUploadCount(volumeName, bucketName);

OmMultipartInfo orphanMpu = createTestMultipartUpload(volumeName, bucketName, "file.txt", owner);

// Age past the threshold.
long oldCreationTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
updateMultipartUploadCreationTime(volumeName, bucketName, "file.txt",
orphanMpu.getUploadID(), oldCreationTime);

// Simulate orphan by removing its open key.
String resolvedOrphanOpenKey = OMMultipartUploadUtils
.getMultipartOpenKey(volumeName, bucketName, "file.txt", orphanMpu.getUploadID(),
metadataManager, OBJECT_STORE);
metadataManager.getOpenKeyTable(OBJECT_STORE).delete(resolvedOrphanOpenKey);

// Rule requires tag match — cannot evaluate without open key.
OmLCRule tagRule = new OmLCRule.Builder()
.setId("abort-by-tag")
.setEnabled(true)
.setFilter(new OmLCFilter.Builder().setTag("env", "test").build())
.setAction(new OmLCAbortIncompleteMultipartUpload.Builder()
.setDaysAfterInitiation(1)
.build())
.build();

createLifecyclePolicy(volumeName, bucketName, OBJECT_STORE, Collections.singletonList(tagRule));

// Wait long enough for the service to run at least once.
Thread.sleep(SERVICE_INTERVAL * 2);

// Orphan MPU must remain: the tag rule cannot evaluate without open key metadata.
String orphanKey = metadataManager.getMultipartKey(volumeName, bucketName,
"file.txt", orphanMpu.getUploadID());
assertNotNull(metadataManager.getMultipartInfoTable().get(orphanKey),
"Orphan MPU should NOT be aborted when only tag-requiring rules exist");

assertEquals(initialMpuCount + 1, getMultipartUploadCount(volumeName, bucketName),
"MPU count should not change");

deleteLifecyclePolicy(volumeName, bucketName);
}

}

/**
Expand Down