diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java index 5310f5b4837e..f54df0efc8fa 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java @@ -1109,55 +1109,81 @@ private void processMultipartUploads(OmBucketInfo bucketInfo, List 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()) { + 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); @@ -1172,33 +1198,16 @@ private void processMultipartUploads(OmBucketInfo bucketInfo, List 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); } /** diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java index 89746fb6c825..e5f041907ee8 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java @@ -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); + } + } /**