You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by iv...@apache.org on 2026/05/02 13:59:53 UTC
(ozone) branch HDDS-8342 updated: HDDS-13474. Support abort incomplete multipart upload action (#10052)
This is an automated email from the ASF dual-hosted git repository.
ivandika3 pushed a commit to branch HDDS-8342
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-8342 by this push:
new f3c6ba4afae HDDS-13474. Support abort incomplete multipart upload action (#10052)
f3c6ba4afae is described below
commit f3c6ba4afae65ecba57a8ef61a6a5a4959c8b3af
Author: Bolin Lin <li...@gmail.com>
AuthorDate: Sat May 2 09:59:46 2026 -0400
HDDS-13474. Support abort incomplete multipart upload action (#10052)
---
.../common/src/main/resources/ozone-default.xml | 9 +
.../ozone/client/OzoneLifecycleConfiguration.java | 32 +-
.../org/apache/hadoop/ozone/om/OMConfigKeys.java | 4 +
.../OmLCAbortIncompleteMultipartUpload.java | 139 ++++++++
.../apache/hadoop/ozone/om/helpers/OmLCAction.java | 1 +
.../apache/hadoop/ozone/om/helpers/OmLCRule.java | 19 +
.../TestOmLCAbortIncompleteMultipartUpload.java | 171 +++++++++
.../hadoop/ozone/om/helpers/TestOmLCRule.java | 105 ++++++
hadoop-ozone/integration-test-s3/pom.xml | 10 +
.../hadoop/ozone/s3/awssdk/OzoneS3SDKTests.java | 11 +
.../ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java | 169 +++++++++
.../ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java | 205 +++++++++++
.../src/main/proto/OmClientProtocol.proto | 5 +
.../ozone/om/service/KeyLifecycleService.java | 386 ++++++++++++++++++---
.../om/service/KeyLifecycleServiceMetrics.java | 25 ++
.../ozone/om/service/TestKeyLifecycleService.java | 318 +++++++++++++++++
.../s3/endpoint/S3LifecycleConfiguration.java | 76 +++-
.../hadoop/ozone/client/OzoneBucketStub.java | 7 +-
.../endpoint/TestS3LifecycleConfigurationGet.java | 74 ++++
.../endpoint/TestS3LifecycleConfigurationPut.java | 86 +++++
20 files changed, 1796 insertions(+), 56 deletions(-)
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index e3b1d74a6f8..9f4e38b0739 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -5095,4 +5095,13 @@
directories will abort.
</description>
</property>
+ <property>
+ <name>ozone.lifecycle.service.mpu.abort.limit.per.task</name>
+ <value>1000</value>
+ <tag>OZONE</tag>
+ <description>Maximum number of multipart upload parts (rounded up to complete uploads)
+ to abort in a single task when processing AbortIncompleteMultipartUpload
+ lifecycle rules.
+ </description>
+ </property>
</configuration>
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneLifecycleConfiguration.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneLifecycleConfiguration.java
index ab4b881e65d..a325f3820d0 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneLifecycleConfiguration.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneLifecycleConfiguration.java
@@ -62,6 +62,21 @@ public Integer getDays() {
}
}
+ /**
+ * A class that encapsulates OzoneLCAbortIncompleteMultipartUpload.
+ */
+ public static class OzoneLCAbortIncompleteMultipartUpload {
+ private final Integer daysAfterInitiation;
+
+ public OzoneLCAbortIncompleteMultipartUpload(Integer daysAfterInitiation) {
+ this.daysAfterInitiation = daysAfterInitiation;
+ }
+
+ public Integer getDaysAfterInitiation() {
+ return daysAfterInitiation;
+ }
+ }
+
/**
* A class that encapsulates {@link org.apache.hadoop.ozone.om.helpers.OmLifecycleRuleAndOperator}.
*/
@@ -120,14 +135,17 @@ public static class OzoneLCRule {
private final String prefix;
private final String status;
private final OzoneLCExpiration expiration;
+ private final OzoneLCAbortIncompleteMultipartUpload abortIncompleteMultipartUpload;
private final OzoneLCFilter filter;
public OzoneLCRule(String id, String prefix, String status,
- OzoneLCExpiration expiration, OzoneLCFilter filter) {
+ OzoneLCExpiration expiration, OzoneLCAbortIncompleteMultipartUpload abortIncompleteMultipartUpload,
+ OzoneLCFilter filter) {
this.id = id;
this.prefix = prefix;
this.status = status;
this.expiration = expiration;
+ this.abortIncompleteMultipartUpload = abortIncompleteMultipartUpload;
this.filter = filter;
}
@@ -147,6 +165,10 @@ public OzoneLCExpiration getExpiration() {
return expiration;
}
+ public OzoneLCAbortIncompleteMultipartUpload getAbortIncompleteMultipartUpload() {
+ return abortIncompleteMultipartUpload;
+ }
+
public OzoneLCFilter getFilter() {
return filter;
}
@@ -181,6 +203,12 @@ public static OzoneLifecycleConfiguration fromOmLifecycleConfiguration(
r.getExpiration().getDays(), r.getExpiration().getDate());
}
+ OzoneLifecycleConfiguration.OzoneLCAbortIncompleteMultipartUpload a = null;
+ if (r.getAbortIncompleteMultipartUpload() != null) {
+ a = new OzoneLifecycleConfiguration.OzoneLCAbortIncompleteMultipartUpload(
+ r.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
OzoneLifecycleConfiguration.OzoneLCFilter f = null;
if (r.getFilter() != null) {
LifecycleAndOperator andOperator = null;
@@ -193,7 +221,7 @@ public static OzoneLifecycleConfiguration fromOmLifecycleConfiguration(
}
rules.add(new OzoneLifecycleConfiguration.OzoneLCRule(r.getId(),
- r.getPrefix(), (r.isEnabled() ? "Enabled" : "Disabled"), e, f));
+ r.getPrefix(), (r.isEnabled() ? "Enabled" : "Disabled"), e, a, f));
}
return new OzoneLifecycleConfiguration(lifecycleConfiguration.getVolume(),
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
index 40f4d59bc24..b5a97376cc1 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java
@@ -188,6 +188,10 @@ public final class OMConfigKeys {
public static final String OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE =
"ozone.lifecycle.service.delete.batch-size";
public static final int OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE_DEFAULT = 1000;
+ // Batch limit for aborting incomplete multipart uploads, based on total part count
+ public static final String OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK =
+ "ozone.lifecycle.service.mpu.abort.limit.per.task";
+ public static final int OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK_DEFAULT = 1000;
public static final String OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT =
"ozone.lifecycle.service.delete.cached.directory.max-count";
public static final long OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT_DEFAULT = 1000000;
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAbortIncompleteMultipartUpload.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAbortIncompleteMultipartUpload.java
new file mode 100644
index 00000000000..a9c06c28f7c
--- /dev/null
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAbortIncompleteMultipartUpload.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.helpers;
+
+import jakarta.annotation.Nullable;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.AbortIncompleteMultipartUpload;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.LifecycleAction;
+
+/**
+ * A class that encapsulates lifecycle rule AbortIncompleteMultipartUpload action.
+ * This class extends OmLCAction and represents the AbortIncompleteMultipartUpload
+ * action type in lifecycle configuration.
+ */
+public final class OmLCAbortIncompleteMultipartUpload implements OmLCAction {
+ private final Integer daysAfterInitiation;
+ private long daysInMilli;
+
+ private OmLCAbortIncompleteMultipartUpload() {
+ throw new UnsupportedOperationException("Default constructor is not supported. Use Builder.");
+ }
+
+ private OmLCAbortIncompleteMultipartUpload(Builder builder) {
+ this.daysAfterInitiation = builder.daysAfterInitiation;
+ }
+
+ @Nullable
+ public Integer getDaysAfterInitiation() {
+ return daysAfterInitiation;
+ }
+
+ /**
+ * Checks if a multipart upload is eligible for abort based on its creation time.
+ *
+ * @param creationTimestamp The creation time of the multipart upload in milliseconds since epoch
+ * @return true if the upload should be aborted, false otherwise
+ */
+ public boolean shouldAbort(long creationTimestamp) {
+ ZonedDateTime now = ZonedDateTime.now(ZoneOffset.UTC);
+ ZonedDateTime dateTime = ZonedDateTime.ofInstant(
+ Instant.ofEpochMilli(creationTimestamp + daysInMilli), ZoneOffset.UTC);
+ return now.isAfter(dateTime);
+ }
+
+ @Override
+ public ActionType getActionType() {
+ return ActionType.ABORT_INCOMPLETE_MULTIPART_UPLOAD;
+ }
+
+ /**
+ * Validates the AbortIncompleteMultipartUpload configuration.
+ * - DaysAfterInitiation must be specified
+ * - DaysAfterInitiation must be a positive number greater than zero
+ *
+ * @param creationTime The creation time of the lifecycle configuration in milliseconds since epoch
+ * @throws OMException if the validation fails
+ */
+ @Override
+ public void valid(long creationTime) throws OMException {
+ if (daysAfterInitiation == null) {
+ throw new OMException("Invalid lifecycle configuration: 'DaysAfterInitiation' " +
+ "must be specified for AbortIncompleteMultipartUpload action.",
+ OMException.ResultCodes.INVALID_REQUEST);
+ }
+
+ if (daysAfterInitiation <= 0) {
+ throw new OMException("'DaysAfterInitiation' for AbortIncompleteMultipartUpload action " +
+ "must be a positive integer greater than zero.",
+ OMException.ResultCodes.INVALID_REQUEST);
+ }
+
+ daysInMilli = TimeUnit.DAYS.toMillis(daysAfterInitiation);
+ }
+
+ @Override
+ public LifecycleAction getProtobuf() {
+ AbortIncompleteMultipartUpload.Builder builder = AbortIncompleteMultipartUpload.newBuilder();
+
+ if (daysAfterInitiation != null) {
+ builder.setDaysAfterInitiation(daysAfterInitiation);
+ }
+
+ return LifecycleAction.newBuilder()
+ .setAbortIncompleteMultipartUpload(builder).build();
+ }
+
+ public static OmLCAbortIncompleteMultipartUpload getFromProtobuf(
+ AbortIncompleteMultipartUpload abortIncompleteMultipartUpload) {
+ OmLCAbortIncompleteMultipartUpload.Builder builder = new Builder();
+
+ if (abortIncompleteMultipartUpload.hasDaysAfterInitiation()) {
+ builder.setDaysAfterInitiation(abortIncompleteMultipartUpload.getDaysAfterInitiation());
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public String toString() {
+ return "OmLCAbortIncompleteMultipartUpload{" +
+ "daysAfterInitiation=" + daysAfterInitiation +
+ '}';
+ }
+
+ /**
+ * Builder of OmLCAbortIncompleteMultipartUpload.
+ */
+ public static class Builder {
+ private Integer daysAfterInitiation = null;
+
+ public Builder setDaysAfterInitiation(int days) {
+ this.daysAfterInitiation = days;
+ return this;
+ }
+
+ public OmLCAbortIncompleteMultipartUpload build() {
+ return new OmLCAbortIncompleteMultipartUpload(this);
+ }
+ }
+}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAction.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAction.java
index 4db13e20743..3d746bd1381 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAction.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCAction.java
@@ -53,6 +53,7 @@ public interface OmLCAction {
*/
enum ActionType {
EXPIRATION,
+ ABORT_INCOMPLETE_MULTIPART_UPLOAD,
// Future action types can be added here (e.g., TRANSITION)
}
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCRule.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCRule.java
index 00875842655..10b627790a0 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCRule.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmLCRule.java
@@ -124,6 +124,21 @@ public OmLCExpiration getExpiration() {
return null;
}
+ /**
+ * Get the AbortIncompleteMultipartUpload action if present.
+ *
+ * @return the AbortIncompleteMultipartUpload action if present, null otherwise
+ */
+ @Nullable
+ public OmLCAbortIncompleteMultipartUpload getAbortIncompleteMultipartUpload() {
+ for (OmLCAction action : actions) {
+ if (action instanceof OmLCAbortIncompleteMultipartUpload) {
+ return (OmLCAbortIncompleteMultipartUpload) action;
+ }
+ }
+ return null;
+ }
+
@Nullable
public OmLCFilter getFilter() {
return filter;
@@ -300,6 +315,10 @@ public static OmLCRule getFromProtobuf(LifecycleRule lifecycleRule, BucketLayout
if (lifecycleAction.hasExpiration()) {
builder.addAction(OmLCExpiration.getFromProtobuf(lifecycleAction.getExpiration()));
}
+ if (lifecycleAction.hasAbortIncompleteMultipartUpload()) {
+ builder.addAction(OmLCAbortIncompleteMultipartUpload.getFromProtobuf(
+ lifecycleAction.getAbortIncompleteMultipartUpload()));
+ }
}
if (lifecycleRule.hasFilter()) {
builder.setFilter(OmLCFilter.getFromProtobuf(lifecycleRule.getFilter(), layout));
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCAbortIncompleteMultipartUpload.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCAbortIncompleteMultipartUpload.java
new file mode 100644
index 00000000000..1a55751c513
--- /dev/null
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCAbortIncompleteMultipartUpload.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.helpers;
+
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_REQUEST;
+import static org.apache.hadoop.ozone.om.helpers.OMLCUtils.assertOMException;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.AbortIncompleteMultipartUpload;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.LifecycleAction;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test OmLCAbortIncompleteMultipartUpload.
+ */
+class TestOmLCAbortIncompleteMultipartUpload {
+
+ @Test
+ public void testCreateValidAbortIncompleteMultipartUpload() {
+ long currentTime = System.currentTimeMillis();
+
+ OmLCAbortIncompleteMultipartUpload.Builder abort1 =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(1);
+ assertDoesNotThrow(() -> abort1.build().valid(currentTime));
+
+ OmLCAbortIncompleteMultipartUpload.Builder abort2 =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(7);
+ assertDoesNotThrow(() -> abort2.build().valid(currentTime));
+
+ OmLCAbortIncompleteMultipartUpload.Builder abort3 =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(365);
+ assertDoesNotThrow(() -> abort3.build().valid(currentTime));
+ }
+
+ @Test
+ public void testCreateInvalidAbortIncompleteMultipartUpload() {
+ long currentTime = System.currentTimeMillis();
+
+ // Null days should fail
+ OmLCAbortIncompleteMultipartUpload.Builder abort1 =
+ new OmLCAbortIncompleteMultipartUpload.Builder();
+ assertOMException(() -> abort1.build().valid(currentTime), INVALID_REQUEST,
+ "must be specified for AbortIncompleteMultipartUpload action");
+
+ // Zero days should fail
+ OmLCAbortIncompleteMultipartUpload.Builder abort2 =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(0);
+ assertOMException(() -> abort2.build().valid(currentTime), INVALID_REQUEST,
+ "must be a positive integer greater than zero");
+
+ // Negative days should fail
+ OmLCAbortIncompleteMultipartUpload.Builder abort3 =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(-1);
+ assertOMException(() -> abort3.build().valid(currentTime), INVALID_REQUEST,
+ "must be a positive integer greater than zero");
+
+ OmLCAbortIncompleteMultipartUpload.Builder abort4 =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(-100);
+ assertOMException(() -> abort4.build().valid(currentTime), INVALID_REQUEST,
+ "must be a positive integer greater than zero");
+ }
+
+ @Test
+ public void testShouldAbort() throws OMException {
+ long currentTime = System.currentTimeMillis();
+
+ // Upload created 10 days ago
+ long uploadCreationTime = currentTime - TimeUnit.DAYS.toMillis(10);
+
+ // Rule: abort after 7 days - should abort
+ OmLCAbortIncompleteMultipartUpload abort7Days =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(7)
+ .build();
+ abort7Days.valid(currentTime);
+ assertTrue(abort7Days.shouldAbort(uploadCreationTime),
+ "Upload created 10 days ago should be aborted with 7-day threshold");
+
+ // Rule: abort after 15 days - should NOT abort
+ OmLCAbortIncompleteMultipartUpload abort15Days =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(15)
+ .build();
+ abort15Days.valid(currentTime);
+ assertFalse(abort15Days.shouldAbort(uploadCreationTime),
+ "Upload created 10 days ago should NOT be aborted with 15-day threshold");
+
+ // Upload created 1 hour ago - should NOT abort
+ long recentUploadTime = currentTime - TimeUnit.HOURS.toMillis(1);
+ OmLCAbortIncompleteMultipartUpload abort1Day =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(1)
+ .build();
+ abort1Day.valid(currentTime);
+ assertFalse(abort1Day.shouldAbort(recentUploadTime),
+ "Upload created 1 hour ago should NOT be aborted with 1-day threshold");
+
+ // Upload created 1 day + 1 second ago - should abort
+ long moreThanOneDay = currentTime - TimeUnit.DAYS.toMillis(1) - TimeUnit.SECONDS.toMillis(1);
+ assertTrue(abort1Day.shouldAbort(moreThanOneDay),
+ "Upload created more than 1 day ago should be aborted");
+ }
+
+ @Test
+ public void testProtobufConversion() throws OMException {
+ long currentTime = System.currentTimeMillis();
+
+ // Create with 7 days
+ OmLCAbortIncompleteMultipartUpload original =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(7)
+ .build();
+
+ // Convert to protobuf
+ LifecycleAction proto = original.getProtobuf();
+ assertTrue(proto.hasAbortIncompleteMultipartUpload(),
+ "Protobuf should have AbortIncompleteMultipartUpload");
+
+ AbortIncompleteMultipartUpload abortProto = proto.getAbortIncompleteMultipartUpload();
+ assertEquals(7, abortProto.getDaysAfterInitiation(),
+ "Days should be preserved in protobuf");
+
+ // Convert back from protobuf
+ OmLCAbortIncompleteMultipartUpload fromProto =
+ OmLCAbortIncompleteMultipartUpload.getFromProtobuf(abortProto);
+ assertEquals(7, fromProto.getDaysAfterInitiation(),
+ "Days should be preserved after protobuf round-trip");
+
+ // Validate the converted object
+ assertDoesNotThrow(() -> fromProto.valid(currentTime),
+ "Object from protobuf should be valid");
+ }
+
+ @Test
+ public void testActionType() {
+ OmLCAbortIncompleteMultipartUpload abort =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(7)
+ .build();
+
+ assertEquals(OmLCAction.ActionType.ABORT_INCOMPLETE_MULTIPART_UPLOAD,
+ abort.getActionType(),
+ "Action type should be ABORT_INCOMPLETE_MULTIPART_UPLOAD");
+ }
+}
diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCRule.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCRule.java
index f81b8afca3a..15e72e5a8c3 100644
--- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCRule.java
+++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/helpers/TestOmLCRule.java
@@ -318,4 +318,109 @@ public void testProtobufConversion() throws OMException {
assertEquals(30, ruleFromProto3.getExpiration().getDays());
assertNull(ruleFromProto3.getFilter());
}
+
+ @Test
+ public void testRuleWithAbortIncompleteMultipartUpload() throws OMException {
+ long currentTime = System.currentTimeMillis();
+
+ // Test rule with only AbortIncompleteMultipartUpload action
+ OmLCAbortIncompleteMultipartUpload abortAction =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(7)
+ .build();
+
+ OmLCRule.Builder rule1 = new OmLCRule.Builder()
+ .setId("abort-incomplete-uploads")
+ .setEnabled(true)
+ .setPrefix("uploads/")
+ .setAction(abortAction);
+
+ OmLCRule builtRule = rule1.build();
+ assertDoesNotThrow(() -> builtRule.valid(BucketLayout.DEFAULT, currentTime));
+ assertNotNull(builtRule.getAbortIncompleteMultipartUpload());
+ assertEquals(7, builtRule.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
+ @Test
+ public void testRuleWithBothExpirationAndAbortActions() throws OMException {
+ long currentTime = System.currentTimeMillis();
+
+ OmLCExpiration expiration = new OmLCExpiration.Builder()
+ .setDays(30)
+ .build();
+
+ OmLCAbortIncompleteMultipartUpload abortAction =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(7)
+ .build();
+
+ OmLCRule.Builder rule = new OmLCRule.Builder()
+ .setId("combined-rule")
+ .setEnabled(true)
+ .setPrefix("temp/")
+ .addAction(expiration)
+ .addAction(abortAction);
+
+ OmLCRule builtRule = rule.build();
+ assertDoesNotThrow(() -> builtRule.valid(BucketLayout.DEFAULT, currentTime));
+ assertNotNull(builtRule.getExpiration());
+ assertNotNull(builtRule.getAbortIncompleteMultipartUpload());
+ assertEquals(30, builtRule.getExpiration().getDays());
+ assertEquals(7, builtRule.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
+ @Test
+ public void testProtobufConversionWithAbortAction() throws OMException {
+ OmLCAbortIncompleteMultipartUpload abortAction =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(14)
+ .build();
+
+ OmLCRule originalRule = new OmLCRule.Builder()
+ .setId("test-abort-rule")
+ .setEnabled(true)
+ .setPrefix("multipart/")
+ .setAction(abortAction)
+ .build();
+
+ LifecycleRule proto = originalRule.getProtobuf();
+
+ OmLCRule ruleFromProto = OmLCRule.getFromProtobuf(proto, BucketLayout.DEFAULT);
+ assertEquals("test-abort-rule", ruleFromProto.getId());
+ assertTrue(ruleFromProto.isEnabled());
+ assertEquals("multipart/", ruleFromProto.getPrefix());
+ assertNotNull(ruleFromProto.getAbortIncompleteMultipartUpload());
+ assertEquals(14, ruleFromProto.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
+ @Test
+ public void testProtobufConversionWithBothActions() throws OMException {
+ OmLCExpiration expiration = new OmLCExpiration.Builder()
+ .setDays(60)
+ .build();
+
+ OmLCAbortIncompleteMultipartUpload abortAction =
+ new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(5)
+ .build();
+
+ OmLCRule originalRule = new OmLCRule.Builder()
+ .setId("combined-actions")
+ .setEnabled(true)
+ .setPrefix("data/")
+ .addAction(expiration)
+ .addAction(abortAction)
+ .build();
+
+ LifecycleRule proto = originalRule.getProtobuf();
+
+ OmLCRule ruleFromProto = OmLCRule.getFromProtobuf(proto, BucketLayout.DEFAULT);
+ assertEquals("combined-actions", ruleFromProto.getId());
+ assertTrue(ruleFromProto.isEnabled());
+ assertEquals("data/", ruleFromProto.getPrefix());
+ assertNotNull(ruleFromProto.getExpiration());
+ assertNotNull(ruleFromProto.getAbortIncompleteMultipartUpload());
+ assertEquals(60, ruleFromProto.getExpiration().getDays());
+ assertEquals(5, ruleFromProto.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
}
diff --git a/hadoop-ozone/integration-test-s3/pom.xml b/hadoop-ozone/integration-test-s3/pom.xml
index 06b5becec17..2a622c3ec1e 100644
--- a/hadoop-ozone/integration-test-s3/pom.xml
+++ b/hadoop-ozone/integration-test-s3/pom.xml
@@ -129,6 +129,16 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.ozone</groupId>
+ <artifactId>ozone-interface-storage</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.ozone</groupId>
+ <artifactId>ozone-manager</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.ozone</groupId>
<artifactId>ozone-mini-cluster</artifactId>
diff --git a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/OzoneS3SDKTests.java b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/OzoneS3SDKTests.java
index 665c9458f3c..0c1ee1c0e33 100644
--- a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/OzoneS3SDKTests.java
+++ b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/OzoneS3SDKTests.java
@@ -17,7 +17,10 @@
package org.apache.hadoop.ozone.s3.awssdk;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.om.OMConfigKeys;
import org.apache.hadoop.ozone.s3.MultiS3GatewayService;
import org.apache.hadoop.ozone.s3.awssdk.v1.AbstractS3SDKV1Tests;
import org.apache.hadoop.ozone.s3.awssdk.v2.AbstractS3SDKV2Tests;
@@ -26,6 +29,14 @@
abstract class OzoneS3SDKTests extends ClusterForTests<MiniOzoneCluster> {
+ @Override
+ protected OzoneConfiguration createOzoneConfig() {
+ OzoneConfiguration conf = createBaseConfiguration();
+ conf.setBoolean(OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED, true);
+ conf.setTimeDuration(OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_INTERVAL, 1, TimeUnit.SECONDS);
+ return conf;
+ }
+
@Override
protected MiniOzoneCluster createCluster() throws Exception {
return newClusterBuilder()
diff --git a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java
index 03aaf4aa610..bb2da9e9fde 100644
--- a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java
+++ b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java
@@ -35,6 +35,7 @@
import com.amazonaws.AmazonServiceException.ErrorType;
import com.amazonaws.HttpMethod;
import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.AbortIncompleteMultipartUpload;
import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
import com.amazonaws.services.s3.model.AccessControlList;
import com.amazonaws.services.s3.model.Bucket;
@@ -103,6 +104,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
@@ -120,15 +122,20 @@
import org.apache.hadoop.ozone.client.OzoneKeyDetails;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
+import org.apache.hadoop.ozone.om.service.KeyLifecycleService;
import org.apache.hadoop.ozone.s3.S3ClientFactory;
import org.apache.hadoop.ozone.s3.awssdk.S3SDKTestUtils;
import org.apache.hadoop.ozone.s3.endpoint.S3Owner;
import org.apache.hadoop.ozone.s3.exception.S3ErrorTable;
import org.apache.hadoop.ozone.s3.util.S3Consts;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ozone.test.GenericTestUtils;
import org.apache.ozone.test.NonHATests;
import org.apache.ozone.test.OzoneTestBase;
import org.junit.jupiter.api.BeforeAll;
@@ -1511,6 +1518,168 @@ public void testS3LifecycleConfigurationGet() {
assertEquals(retrievedConfig.getRules().size(), configFromRequest.getRules().size());
}
+ @Test
+ public void testGetLifecycleWithAbortIncompleteMultipartUpload() {
+ final String bucketName = getBucketName();
+ s3Client.createBucket(bucketName);
+
+ BucketLifecycleConfiguration configuration = new BucketLifecycleConfiguration();
+ List<BucketLifecycleConfiguration.Rule> rules = new ArrayList<>();
+
+ BucketLifecycleConfiguration.Rule rule1 = new BucketLifecycleConfiguration.Rule()
+ .withId("abort-incomplete-mpu-with-prefix")
+ .withPrefix("uploads/")
+ .withStatus(BucketLifecycleConfiguration.ENABLED);
+ rule1.setAbortIncompleteMultipartUpload(
+ new AbortIncompleteMultipartUpload().withDaysAfterInitiation(7));
+
+ BucketLifecycleConfiguration.Rule rule2 = new BucketLifecycleConfiguration.Rule()
+ .withId("abort-incomplete-mpu-temp")
+ .withPrefix("temp/")
+ .withStatus(BucketLifecycleConfiguration.ENABLED);
+ rule2.setAbortIncompleteMultipartUpload(
+ new AbortIncompleteMultipartUpload().withDaysAfterInitiation(3));
+
+ BucketLifecycleConfiguration.Rule rule3 = new BucketLifecycleConfiguration.Rule()
+ .withId("abort-incomplete-mpu-no-prefix")
+ .withPrefix("")
+ .withStatus(BucketLifecycleConfiguration.ENABLED);
+ rule3.setAbortIncompleteMultipartUpload(
+ new AbortIncompleteMultipartUpload().withDaysAfterInitiation(30));
+
+ rules.add(rule1);
+ rules.add(rule2);
+ rules.add(rule3);
+ configuration.setRules(rules);
+
+ // Set lifecycle configuration
+ s3Client.setBucketLifecycleConfiguration(bucketName, configuration);
+
+ // Get and verify the configuration
+ BucketLifecycleConfiguration retrievedConfig =
+ s3Client.getBucketLifecycleConfiguration(bucketName);
+
+ assertEquals(3, retrievedConfig.getRules().size());
+
+ BucketLifecycleConfiguration.Rule retrievedRule1 = retrievedConfig.getRules().get(0);
+ assertEquals("abort-incomplete-mpu-with-prefix", retrievedRule1.getId());
+ assertEquals("uploads/", retrievedRule1.getPrefix());
+ assertEquals(BucketLifecycleConfiguration.ENABLED, retrievedRule1.getStatus());
+ assertEquals(7, retrievedRule1.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+
+ BucketLifecycleConfiguration.Rule retrievedRule2 = retrievedConfig.getRules().get(1);
+ assertEquals("abort-incomplete-mpu-temp", retrievedRule2.getId());
+ assertEquals("temp/", retrievedRule2.getPrefix());
+ assertEquals(BucketLifecycleConfiguration.ENABLED, retrievedRule2.getStatus());
+ assertEquals(3, retrievedRule2.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+
+ BucketLifecycleConfiguration.Rule retrievedRule3 = retrievedConfig.getRules().get(2);
+ assertEquals("abort-incomplete-mpu-no-prefix", retrievedRule3.getId());
+ assertEquals("", retrievedRule3.getPrefix());
+ assertEquals(BucketLifecycleConfiguration.ENABLED, retrievedRule3.getStatus());
+ assertEquals(30, retrievedRule3.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
+ /**
+ * End-to-end test verifying that KeyLifecycleService correctly aborts
+ * incomplete multipart uploads based on lifecycle configuration.
+ */
+ @Test
+ void testAbortIncompleteMultipartUploadE2E() throws Exception {
+ OzoneManager ozoneManager = cluster().getOzoneManager();
+ KeyLifecycleService lifecycleService = ozoneManager.getKeyManager().getKeyLifecycleService();
+ if (lifecycleService == null) {
+ return;
+ }
+
+ final String bucketName = getBucketName();
+ s3Client.createBucket(bucketName);
+
+ String s3VolumeName;
+ try (OzoneClient ozoneClient = cluster().newClient()) {
+ s3VolumeName = ozoneClient.getObjectStore().getS3Volume().getName();
+ }
+
+ OMMetadataManager metadataManager = ozoneManager.getMetadataManager();
+
+ // Create 3 MPUs
+ String matchingOldKey = "temp/old-file.txt";
+ InitiateMultipartUploadResult mpu1 = s3Client.initiateMultipartUpload(
+ new InitiateMultipartUploadRequest(bucketName, matchingOldKey));
+
+ String nonMatchingOldKey = "permanent/old-file.txt";
+ InitiateMultipartUploadResult mpu2 = s3Client.initiateMultipartUpload(
+ new InitiateMultipartUploadRequest(bucketName, nonMatchingOldKey));
+
+ String matchingRecentKey = "temp/recent-file.txt";
+ s3Client.initiateMultipartUpload(new InitiateMultipartUploadRequest(bucketName, matchingRecentKey));
+
+ MultipartUploadListing listBefore = s3Client.listMultipartUploads(
+ new ListMultipartUploadsRequest(bucketName));
+ assertEquals(3, listBefore.getMultipartUploads().size());
+
+ // Backdate 2 MPUs to 2 days ago
+ long oldCreationTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
+ updateMpuCreationTime(metadataManager, s3VolumeName, bucketName,
+ matchingOldKey, mpu1.getUploadId(), oldCreationTime);
+ updateMpuCreationTime(metadataManager, s3VolumeName, bucketName,
+ nonMatchingOldKey, mpu2.getUploadId(), oldCreationTime);
+
+ // Set lifecycle rule: abort MPUs with prefix "temp/" after 1 day
+ BucketLifecycleConfiguration.Rule rule = new BucketLifecycleConfiguration.Rule()
+ .withId("abort-temp-uploads")
+ .withPrefix("temp/")
+ .withStatus(BucketLifecycleConfiguration.ENABLED);
+ rule.setAbortIncompleteMultipartUpload(
+ new AbortIncompleteMultipartUpload().withDaysAfterInitiation(1));
+
+ BucketLifecycleConfiguration config = new BucketLifecycleConfiguration();
+ config.setRules(Collections.singletonList(rule));
+ s3Client.setBucketLifecycleConfiguration(bucketName, config);
+
+ // Trigger lifecycle service
+ lifecycleService.runPeriodicalTaskNow();
+
+ // Wait for abort
+ GenericTestUtils.waitFor(() -> {
+ MultipartUploadListing listing = s3Client.listMultipartUploads(
+ new ListMultipartUploadsRequest(bucketName));
+ return listing.getMultipartUploads().size() == 2;
+ }, 500, 30000);
+
+ // Verify results
+ MultipartUploadListing listAfter = s3Client.listMultipartUploads(
+ new ListMultipartUploadsRequest(bucketName));
+ List<String> remainingKeys = listAfter.getMultipartUploads().stream()
+ .map(MultipartUpload::getKey)
+ .collect(Collectors.toList());
+
+ assertFalse(remainingKeys.contains(matchingOldKey), "Old MPU with matching prefix should be aborted");
+ assertTrue(remainingKeys.contains(nonMatchingOldKey), "Old MPU with non-matching prefix should remain");
+ assertTrue(remainingKeys.contains(matchingRecentKey), "Recent MPU should remain");
+ }
+
+ private void updateMpuCreationTime(OMMetadataManager metadataManager,
+ String volumeName, String bucketName, String keyName,
+ String uploadId, long newCreationTime) throws Exception {
+ String multipartKey = metadataManager.getMultipartKey(volumeName, bucketName, keyName, uploadId);
+ OmMultipartKeyInfo existingInfo = metadataManager.getMultipartInfoTable().get(multipartKey);
+ if (existingInfo == null) {
+ throw new RuntimeException("Multipart key info not found: " + multipartKey);
+ }
+
+ OmMultipartKeyInfo updatedInfo = new OmMultipartKeyInfo.Builder()
+ .setUploadID(existingInfo.getUploadID())
+ .setCreationTime(newCreationTime)
+ .setReplicationConfig(existingInfo.getReplicationConfig())
+ .setObjectID(existingInfo.getObjectID())
+ .setUpdateID(existingInfo.getUpdateID())
+ .setParentID(existingInfo.getParentID())
+ .build();
+
+ metadataManager.getMultipartInfoTable().put(multipartKey, updatedInfo);
+ }
+
@Nested
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class PresignedUrlTests {
diff --git a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java
index 51d0d7bbd20..8f50c7c8abc 100644
--- a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java
+++ b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java
@@ -52,6 +52,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.xml.bind.DatatypeConverter;
@@ -70,12 +71,17 @@
import org.apache.hadoop.ozone.client.OzoneKeyDetails;
import org.apache.hadoop.ozone.client.OzoneVolume;
import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.service.KeyLifecycleService;
import org.apache.hadoop.ozone.s3.S3ClientFactory;
import org.apache.hadoop.ozone.s3.awssdk.S3SDKTestUtils;
import org.apache.hadoop.ozone.s3.endpoint.S3Owner;
import org.apache.hadoop.ozone.s3.exception.S3ErrorTable;
import org.apache.hadoop.ozone.s3.util.S3Consts;
import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.ozone.test.GenericTestUtils;
import org.apache.ozone.test.NonHATests;
import org.apache.ozone.test.OzoneTestBase;
import org.junit.jupiter.api.AfterAll;
@@ -102,7 +108,9 @@
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.AbortIncompleteMultipartUpload;
import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
+import software.amazon.awssdk.services.s3.model.BucketLifecycleConfiguration;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
@@ -116,15 +124,21 @@
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectTaggingRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
+import software.amazon.awssdk.services.s3.model.ExpirationStatus;
import software.amazon.awssdk.services.s3.model.GetBucketAclRequest;
+import software.amazon.awssdk.services.s3.model.GetBucketLifecycleConfigurationResponse;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.services.s3.model.GetObjectTaggingRequest;
import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.LifecycleRule;
+import software.amazon.awssdk.services.s3.model.LifecycleRuleAndOperator;
+import software.amazon.awssdk.services.s3.model.LifecycleRuleFilter;
import software.amazon.awssdk.services.s3.model.ListBucketsResponse;
import software.amazon.awssdk.services.s3.model.ListMultipartUploadsRequest;
+import software.amazon.awssdk.services.s3.model.ListMultipartUploadsResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsRequest;
import software.amazon.awssdk.services.s3.model.ListObjectsResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
@@ -1507,6 +1521,197 @@ public void testReadSnapshotDirectoryUsingS3SDK() throws Exception {
assertEquals(content, snapshotResponse.asUtf8String());
}
+ @Test
+ public void testGetLifecycleWithAbortIncompleteMultipartUpload() {
+ final String bucketName = getBucketName();
+ s3Client.createBucket(b -> b.bucket(bucketName));
+
+ LifecycleRule rule1 = LifecycleRule.builder()
+ .id("abort-incomplete-mpu-with-prefix")
+ .prefix("uploads/")
+ .status(ExpirationStatus.ENABLED)
+ .abortIncompleteMultipartUpload(AbortIncompleteMultipartUpload.builder()
+ .daysAfterInitiation(7)
+ .build())
+ .build();
+
+ LifecycleRule rule2 = LifecycleRule.builder()
+ .id("abort-incomplete-mpu-with-tag")
+ .filter(LifecycleRuleFilter.builder()
+ .tag(Tag.builder().key("env").value("dev").build())
+ .build())
+ .status(ExpirationStatus.ENABLED)
+ .abortIncompleteMultipartUpload(AbortIncompleteMultipartUpload.builder()
+ .daysAfterInitiation(14)
+ .build())
+ .build();
+
+ LifecycleRule rule3 = LifecycleRule.builder()
+ .id("abort-incomplete-mpu-with-and-operator")
+ .filter(LifecycleRuleFilter.builder()
+ .and(LifecycleRuleAndOperator.builder()
+ .prefix("temp/")
+ .tags(Tag.builder().key("type").value("temporary").build())
+ .build())
+ .build())
+ .status(ExpirationStatus.ENABLED)
+ .abortIncompleteMultipartUpload(AbortIncompleteMultipartUpload.builder()
+ .daysAfterInitiation(3)
+ .build())
+ .build();
+
+ LifecycleRule rule4 = LifecycleRule.builder()
+ .id("abort-incomplete-mpu-no-filter")
+ .prefix("")
+ .status(ExpirationStatus.ENABLED)
+ .abortIncompleteMultipartUpload(AbortIncompleteMultipartUpload.builder()
+ .daysAfterInitiation(30)
+ .build())
+ .build();
+
+ BucketLifecycleConfiguration configuration = BucketLifecycleConfiguration.builder()
+ .rules(rule1, rule2, rule3, rule4)
+ .build();
+
+ s3Client.putBucketLifecycleConfiguration(b -> b
+ .bucket(bucketName)
+ .lifecycleConfiguration(configuration));
+
+ GetBucketLifecycleConfigurationResponse response =
+ s3Client.getBucketLifecycleConfiguration(b -> b.bucket(bucketName));
+
+ List<LifecycleRule> rules = response.rules();
+ assertEquals(4, rules.size());
+
+ LifecycleRule retrievedRule1 = rules.get(0);
+ assertEquals("abort-incomplete-mpu-with-prefix", retrievedRule1.id());
+ assertEquals("uploads/", retrievedRule1.prefix());
+ assertEquals(ExpirationStatus.ENABLED, retrievedRule1.status());
+ assertEquals(7, retrievedRule1.abortIncompleteMultipartUpload().daysAfterInitiation());
+
+ LifecycleRule retrievedRule2 = rules.get(1);
+ assertEquals("abort-incomplete-mpu-with-tag", retrievedRule2.id());
+ assertEquals(ExpirationStatus.ENABLED, retrievedRule2.status());
+ assertEquals(14, retrievedRule2.abortIncompleteMultipartUpload().daysAfterInitiation());
+ assertEquals("env", retrievedRule2.filter().tag().key());
+ assertEquals("dev", retrievedRule2.filter().tag().value());
+
+ LifecycleRule retrievedRule3 = rules.get(2);
+ assertEquals("abort-incomplete-mpu-with-and-operator", retrievedRule3.id());
+ assertEquals(ExpirationStatus.ENABLED, retrievedRule3.status());
+ assertEquals(3, retrievedRule3.abortIncompleteMultipartUpload().daysAfterInitiation());
+ assertEquals("temp/", retrievedRule3.filter().and().prefix());
+ assertEquals(1, retrievedRule3.filter().and().tags().size());
+ Tag andTag = retrievedRule3.filter().and().tags().get(0);
+ assertEquals("type", andTag.key());
+ assertEquals("temporary", andTag.value());
+
+ LifecycleRule retrievedRule4 = rules.get(3);
+ assertEquals("abort-incomplete-mpu-no-filter", retrievedRule4.id());
+ assertEquals("", retrievedRule4.prefix());
+ assertEquals(ExpirationStatus.ENABLED, retrievedRule4.status());
+ assertEquals(30, retrievedRule4.abortIncompleteMultipartUpload().daysAfterInitiation());
+ }
+
+ /**
+ * End-to-end test verifying that KeyLifecycleService correctly aborts
+ * incomplete multipart uploads based on lifecycle configuration.
+ */
+ @Test
+ void testAbortIncompleteMultipartUploadE2E() throws Exception {
+ OzoneManager ozoneManager = cluster().getOzoneManager();
+ KeyLifecycleService lifecycleService = ozoneManager.getKeyManager().getKeyLifecycleService();
+ if (lifecycleService == null) {
+ return;
+ }
+
+ final String bucketName = getBucketName();
+ s3Client.createBucket(b -> b.bucket(bucketName));
+
+ String s3VolumeName;
+ try (OzoneClient ozoneClient = cluster().newClient()) {
+ s3VolumeName = ozoneClient.getObjectStore().getS3Volume().getName();
+ }
+
+ OMMetadataManager metadataManager = ozoneManager.getMetadataManager();
+
+ // Create 3 MPUs
+ String matchingOldKey = "temp/old-file.txt";
+ CreateMultipartUploadResponse mpu1 = s3Client.createMultipartUpload(
+ b -> b.bucket(bucketName).key(matchingOldKey));
+
+ String nonMatchingOldKey = "permanent/old-file.txt";
+ CreateMultipartUploadResponse mpu2 = s3Client.createMultipartUpload(
+ b -> b.bucket(bucketName).key(nonMatchingOldKey));
+
+ String matchingRecentKey = "temp/recent-file.txt";
+ s3Client.createMultipartUpload(b -> b.bucket(bucketName).key(matchingRecentKey));
+
+ assertEquals(3, s3Client.listMultipartUploads(b -> b.bucket(bucketName)).uploads().size());
+
+ // Backdate 2 MPUs to 2 days ago
+ long oldCreationTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
+ updateMpuCreationTime(metadataManager, s3VolumeName, bucketName,
+ matchingOldKey, mpu1.uploadId(), oldCreationTime);
+ updateMpuCreationTime(metadataManager, s3VolumeName, bucketName,
+ nonMatchingOldKey, mpu2.uploadId(), oldCreationTime);
+
+ // Set lifecycle rule: abort MPUs with prefix "temp/" after 1 day
+ LifecycleRule rule = LifecycleRule.builder()
+ .id("abort-temp-uploads")
+ .prefix("temp/")
+ .status(ExpirationStatus.ENABLED)
+ .abortIncompleteMultipartUpload(AbortIncompleteMultipartUpload.builder()
+ .daysAfterInitiation(1)
+ .build())
+ .build();
+
+ s3Client.putBucketLifecycleConfiguration(b -> b
+ .bucket(bucketName)
+ .lifecycleConfiguration(BucketLifecycleConfiguration.builder()
+ .rules(rule)
+ .build()));
+
+ // Trigger lifecycle service
+ lifecycleService.runPeriodicalTaskNow();
+
+ // Wait for abort
+ GenericTestUtils.waitFor(() -> {
+ return s3Client.listMultipartUploads(b -> b.bucket(bucketName)).uploads().size() == 2;
+ }, 500, 30000);
+
+ // Verify results
+ ListMultipartUploadsResponse listAfter = s3Client.listMultipartUploads(b -> b.bucket(bucketName));
+ List<String> remainingKeys = listAfter.uploads().stream()
+ .map(u -> u.key())
+ .collect(Collectors.toList());
+
+ assertFalse(remainingKeys.contains(matchingOldKey), "Old MPU with matching prefix should be aborted");
+ assertTrue(remainingKeys.contains(nonMatchingOldKey), "Old MPU with non-matching prefix should remain");
+ assertTrue(remainingKeys.contains(matchingRecentKey), "Recent MPU should remain");
+ }
+
+ private void updateMpuCreationTime(OMMetadataManager metadataManager,
+ String volumeName, String bucketName, String keyName,
+ String uploadId, long newCreationTime) throws Exception {
+ String multipartKey = metadataManager.getMultipartKey(volumeName, bucketName, keyName, uploadId);
+ OmMultipartKeyInfo existingInfo = metadataManager.getMultipartInfoTable().get(multipartKey);
+ if (existingInfo == null) {
+ throw new RuntimeException("Multipart key info not found: " + multipartKey);
+ }
+
+ OmMultipartKeyInfo updatedInfo = new OmMultipartKeyInfo.Builder()
+ .setUploadID(existingInfo.getUploadID())
+ .setCreationTime(newCreationTime)
+ .setReplicationConfig(existingInfo.getReplicationConfig())
+ .setObjectID(existingInfo.getObjectID())
+ .setUpdateID(existingInfo.getUpdateID())
+ .setParentID(existingInfo.getParentID())
+ .build();
+
+ metadataManager.getMultipartInfoTable().put(multipartKey, updatedInfo);
+ }
+
private String getBucketName() {
return getBucketName("");
}
diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
index bb37a08bcef..19828030d64 100644
--- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
+++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto
@@ -2534,6 +2534,10 @@ message LifecycleExpiration {
optional string date = 2;
}
+message AbortIncompleteMultipartUpload {
+ optional uint32 daysAfterInitiation = 1;
+}
+
message LifecycleRule {
required string id = 1;
required bool enabled = 2;
@@ -2544,6 +2548,7 @@ message LifecycleRule {
message LifecycleAction {
optional LifecycleExpiration expiration = 1;
+ optional AbortIncompleteMultipartUpload abortIncompleteMultipartUpload = 2;
}
message LifecycleConfiguration {
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 3666401da16..dd37430dccf 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
@@ -28,6 +28,8 @@
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED_DEFAULT;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED_DEFAULT;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK_DEFAULT;
import static org.apache.hadoop.ozone.om.helpers.BucketLayout.OBJECT_STORE;
import com.google.common.annotations.VisibleForTesting;
@@ -37,6 +39,7 @@
import java.io.IOException;
import java.nio.file.Paths;
import java.security.PrivilegedExceptionAction;
+import java.time.Instant;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
@@ -77,11 +80,15 @@
import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmLCFilter;
import org.apache.hadoop.ozone.om.helpers.OmLCRule;
import org.apache.hadoop.ozone.om.helpers.OmLifecycleConfiguration;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartUpload;
import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils;
import org.apache.hadoop.ozone.om.request.OMClientRequest;
+import org.apache.hadoop.ozone.om.request.util.OMMultipartUploadUtils;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateDirectoryRequest;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteKeyArgs;
@@ -108,6 +115,7 @@ public class KeyLifecycleService extends BackgroundService {
private final OzoneManager ozoneManager;
private int keyDeleteBatchSize;
private int listMaxSize;
+ private int mpuAbortLimitPerTask;
private long cachedDirMaxCount;
private final AtomicBoolean suspended;
private final AtomicBoolean isServiceEnabled;
@@ -134,6 +142,10 @@ public KeyLifecycleService(OzoneManager ozoneManager,
Preconditions.checkArgument(keyDeleteBatchSize > 0,
OZONE_KEY_LIFECYCLE_SERVICE_DELETE_BATCH_SIZE + " should be a positive value.");
this.listMaxSize = keyDeleteBatchSize >= 10000 ? keyDeleteBatchSize : 10000;
+ this.mpuAbortLimitPerTask = conf.getInt(OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK,
+ OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK_DEFAULT);
+ Preconditions.checkArgument(mpuAbortLimitPerTask > 0,
+ OZONE_KEY_LIFECYCLE_SERVICE_MPU_ABORT_LIMIT_PER_TASK + " should be a positive value.");
this.cachedDirMaxCount = conf.getLong(OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT,
OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT_DEFAULT);
this.suspended = new AtomicBoolean(false);
@@ -264,6 +276,8 @@ public final class LifecycleActionTask implements BackgroundTask {
private long numKeyRenamed = 0;
private long sizeKeyRenamed = 0;
private long numDirRenamed = 0;
+ private long numMultipartUploadIterated = 0;
+ private long numMultipartUploadAborted = 0;
public LifecycleActionTask(OmLifecycleConfiguration lcConfig) {
this.policy = lcConfig;
@@ -309,63 +323,74 @@ public BackgroundTaskResult call() {
// remove disabled rules
List<OmLCRule> ruleList = originRuleList.stream().filter(r -> r.isEnabled()).collect(Collectors.toList());
- // scan file or key table for evaluate rules against files or keys
- LimitedExpiredObjectList expiredKeyList = new LimitedExpiredObjectList(listMaxSize);
- LimitedExpiredObjectList expiredDirList = new LimitedExpiredObjectList(listMaxSize);
- Table<String, OmKeyInfo> keyTable = omMetadataManager.getKeyTable(bucket.getBucketLayout());
- /**
- * Filter treatment.
- * "" - all objects
- * "/" - if it's OBS/Legacy, means keys starting with "/"; If it's FSO, not supported
- * "/key" - if it's OBS/Legacy, means keys starting with "/key", "/" is literally "/";
- * If it's FSO, means keys or dirs starting with "key", "/" will be treated as separator mark.
- * "key" - if it's OBS/Legacy, means keys starting with "key";
- * if it's FSO, means keys for dirs starting with "key" too.
- * "dir/" - if it's OBS/Legacy, means keys starting with "dir/";
- * - if it's FSO, means keys/dirs under directory "dir", doesn't include directory "dir" itself.
- * - For FSO bucket, as directory ModificationTime will not be updated when any of its child key/subdir
- * changes, so remember to add the tailing slash "/" when configure prefix, otherwise the whole
- * directory will be expired and deleted once its ModificationTime meats the condition.
- */
- if (bucket.getBucketLayout() == BucketLayout.FILE_SYSTEM_OPTIMIZED) {
- OmVolumeArgs volume;
- try {
- volume = omMetadataManager.getVolumeTable().get(omMetadataManager.getVolumeKey(bucket.getVolumeName()));
- if (volume == null) {
- LOG.warn("Volume {} cannot be found, might be deleted during this task's execution",
- bucket.getVolumeName());
+ List<OmLCRule> expirationRules = ruleList.stream()
+ .filter(r -> r.getExpiration() != null)
+ .collect(Collectors.toList());
+ List<OmLCRule> mpuRules = ruleList.stream()
+ .filter(r -> r.getAbortIncompleteMultipartUpload() != null)
+ .collect(Collectors.toList());
+
+ if (!expirationRules.isEmpty()) {
+ LimitedExpiredObjectList expiredKeyList = new LimitedExpiredObjectList(listMaxSize);
+ LimitedExpiredObjectList expiredDirList = new LimitedExpiredObjectList(listMaxSize);
+ Table<String, OmKeyInfo> keyTable = omMetadataManager.getKeyTable(bucket.getBucketLayout());
+ /**
+ * Filter treatment.
+ * "" - all objects
+ * "/" - if it's OBS/Legacy, means keys starting with "/"; If it's FSO, not supported
+ * "/key" - if it's OBS/Legacy, means keys starting with "/key", "/" is literally "/";
+ * If it's FSO, means keys or dirs starting with "key", "/" will be treated as separator mark.
+ * "key" - if it's OBS/Legacy, means keys starting with "key";
+ * if it's FSO, means keys for dirs starting with "key" too.
+ * "dir/" - if it's OBS/Legacy, means keys starting with "dir/";
+ * - if it's FSO, means keys/dirs under directory "dir", doesn't include directory "dir" itself.
+ * - For FSO bucket, as directory ModificationTime will not be updated when any of its child key/subdir
+ * changes, so remember to add the tailing slash "/" when configure prefix, otherwise the whole
+ * directory will be expired and deleted once its ModificationTime meats the condition.
+ */
+ if (bucket.getBucketLayout() == BucketLayout.FILE_SYSTEM_OPTIMIZED) {
+ OmVolumeArgs volume;
+ try {
+ volume = omMetadataManager.getVolumeTable().get(omMetadataManager.getVolumeKey(bucket.getVolumeName()));
+ if (volume == null) {
+ LOG.warn("Volume {} cannot be found, might be deleted during this task's execution",
+ bucket.getVolumeName());
+ onFailure(bucketKey);
+ return result;
+ }
+ } catch (IOException e) {
+ LOG.warn("Failed to get volume {}", bucket.getVolumeName(), e);
onFailure(bucketKey);
return result;
}
- } catch (IOException e) {
- LOG.warn("Failed to get volume {}", bucket.getVolumeName(), e);
- onFailure(bucketKey);
- return result;
+ evaluateFSOBucket(volume, bucket, bucketKey, keyTable, expirationRules, expiredKeyList, expiredDirList);
+ } else {
+ // use bucket name as key iterator prefix
+ evaluateBucket(bucket, keyTable, expirationRules, expiredKeyList);
}
- evaluateFSOBucket(volume, bucket, bucketKey, keyTable, ruleList, expiredKeyList, expiredDirList);
- } else {
- // use bucket name as key iterator prefix
- evaluateBucket(bucket, keyTable, ruleList, expiredKeyList);
- }
- if (expiredKeyList.isEmpty() && expiredDirList.isEmpty()) {
- LOG.info("No expired keys/dirs found/remained for bucket {}", bucketKey);
- onSuccess(bucketKey);
- return result;
- }
+ if (expiredKeyList.isEmpty() && expiredDirList.isEmpty()) {
+ LOG.info("No expired keys/dirs found/remained for bucket {}", bucketKey);
+ } else {
+ LOG.info("{} expired keys and {} expired dirs found and remained for bucket {}",
+ expiredKeyList.size(), expiredDirList.size(), bucketKey);
- LOG.info("{} expired keys and {} expired dirs found and remained for bucket {}",
- expiredKeyList.size(), expiredDirList.size(), bucketKey);
+ // If trash is enabled, move files to trash, instead of send delete requests.
+ // OBS bucket doesn't support trash.
+ if (bucket.getBucketLayout() == OBJECT_STORE) {
+ sendDeleteKeysRequestAndClearList(bucket.getVolumeName(), bucket.getBucketName(), expiredKeyList, false);
+ } else {
+ // handle keys first, then directories
+ handleAndClearFullList(bucket, expiredKeyList, false);
+ handleAndClearFullList(bucket, expiredDirList, true);
+ }
+ }
+ }
- // If trash is enabled, move files to trash, instead of send delete requests.
- // OBS bucket doesn't support trash.
- if (bucket.getBucketLayout() == OBJECT_STORE) {
- sendDeleteKeysRequestAndClearList(bucket.getVolumeName(), bucket.getBucketName(), expiredKeyList, false);
- } else {
- // handle keys first, then directories
- handleAndClearFullList(bucket, expiredKeyList, false);
- handleAndClearFullList(bucket, expiredDirList, true);
+ if (!mpuRules.isEmpty()) {
+ processMultipartUploads(bucket, mpuRules);
}
+
onSuccess(bucketKey);
}
@@ -743,6 +768,198 @@ private void evaluateBucket(OmBucketInfo bucketInfo,
}
}
+ /**
+ * Process AbortIncompleteMultipartUpload actions for incomplete multipart uploads.
+ * Iterates through the multipartInfoTable and aborts uploads that match the rule criteria
+ * and have exceeded the configured days after initiation.
+ *
+ * @param bucketInfo the bucket information
+ * @param ruleList list of lifecycle rules with AbortIncompleteMultipartUpload action
+ */
+ private void processMultipartUploads(OmBucketInfo bucketInfo, List<OmLCRule> ruleList) {
+ String volumeName = bucketInfo.getVolumeName();
+ String bucketName = bucketInfo.getBucketName();
+ String bucketPrefix = omMetadataManager.getBucketKeyPrefix(volumeName, bucketName);
+
+ LOG.debug("Processing AbortIncompleteMultipartUpload actions for bucket {}/{}", volumeName, bucketName);
+
+ PartCountLimitedList expiredUploads = new PartCountLimitedList(mpuAbortLimitPerTask);
+ try (TableIterator<String, ? extends Table.KeyValue<String, OmMultipartKeyInfo>> mpuIterator =
+ omMetadataManager.getMultipartInfoTable().iterator(bucketPrefix)) {
+ while (mpuIterator.hasNext()) {
+ if (!shouldRun()) {
+ LOG.info("KeyLifecycleService is suspended or disabled. " +
+ "Stopping multipart upload processing for bucket {}.", bucketName);
+ return;
+ }
+
+ Table.KeyValue<String, OmMultipartKeyInfo> entry = mpuIterator.next();
+ OmMultipartKeyInfo mpuKeyInfo = entry.getValue();
+ numMultipartUploadIterated++;
+
+ OmMultipartUpload upload;
+ try {
+ upload = OmMultipartUpload.from(entry.getKey());
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Failed to parse multipart upload key {} in bucket {}/{}, skipping",
+ entry.getKey(), volumeName, bucketName, e);
+ continue;
+ }
+
+ 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;
+ }
+
+ 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);
+ }
+
+ // Get part count for this MPU (at least 1 even if no parts uploaded yet)
+ int partCount = Math.max(1, mpuKeyInfo.getPartKeyInfoMap().size());
+ expiredUploads.add(upload, partCount);
+ LOG.debug("Multipart upload {}/{}/{} with uploadId {} ({} parts) will be aborted",
+ volumeName, bucketName, keyName, upload.getUploadId(), partCount);
+ break;
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOG.warn("Failed to iterate multipartInfoTable for bucket {}/{}", volumeName, bucketName, e);
+ return;
+ }
+
+ if (!expiredUploads.isEmpty()) {
+ LOG.info("{} expired multipart uploads ({} parts) remaining for bucket {}/{}",
+ expiredUploads.size(), expiredUploads.getPartCount(), volumeName, bucketName);
+ abortExpiredMultipartUploadsAndClear(bucketInfo, expiredUploads);
+ }
+ }
+
+ /**
+ * 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
+ */
+ private boolean shouldAbortUpload(OmKeyInfo openKeyInfo, 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;
+ }
+
+ /**
+ * Abort expired multipart uploads by sending an abort request.
+ *
+ * @param bucketInfo the bucket information
+ * @param expiredUploads list of expired multipart uploads to abort
+ */
+ private void abortExpiredMultipartUploads(OmBucketInfo bucketInfo, List<OmMultipartUpload> expiredUploads) {
+ String volumeName = bucketInfo.getVolumeName();
+ String bucketName = bucketInfo.getBucketName();
+
+ List<OzoneManagerProtocolProtos.ExpiredMultipartUploadInfo> expiredMPUInfoList = expiredUploads.stream()
+ .map(upload -> OzoneManagerProtocolProtos.ExpiredMultipartUploadInfo.newBuilder()
+ .setName(upload.getDbKey())
+ .build())
+ .collect(Collectors.toList());
+
+ OzoneManagerProtocolProtos.ExpiredMultipartUploadsBucket expiredMPUBucket =
+ OzoneManagerProtocolProtos.ExpiredMultipartUploadsBucket.newBuilder()
+ .setVolumeName(volumeName)
+ .setBucketName(bucketName)
+ .addAllMultipartUploads(expiredMPUInfoList)
+ .build();
+
+ OzoneManagerProtocolProtos.MultipartUploadsExpiredAbortRequest abortRequest =
+ OzoneManagerProtocolProtos.MultipartUploadsExpiredAbortRequest.newBuilder()
+ .addExpiredMultipartUploadsPerBucket(expiredMPUBucket)
+ .build();
+
+ OMRequest omRequest = OMRequest.newBuilder()
+ .setCmdType(OzoneManagerProtocolProtos.Type.AbortExpiredMultiPartUploads)
+ .setMultipartUploadsExpiredAbortRequest(abortRequest)
+ .setVersion(ClientVersion.CURRENT_VERSION)
+ .setClientId(clientId.toString())
+ .build();
+
+ try {
+ long startTime = System.nanoTime();
+ OzoneManagerProtocolProtos.OMResponse response = OzoneManagerRatisUtils.submitRequest(
+ getOzoneManager(), omRequest, clientId, callId.getAndIncrement());
+ long endTime = System.nanoTime();
+
+ if (response != null) {
+ if (response.getSuccess()) {
+ numMultipartUploadAborted += expiredUploads.size();
+
+ LOG.info("Successfully aborted {} multipart uploads for bucket {}/{} in {} ns",
+ expiredUploads.size(), volumeName, bucketName, endTime - startTime);
+ } else {
+ LOG.error("Failed to abort multipart uploads for bucket {}/{}: {}",
+ volumeName, bucketName, response.getMessage());
+ }
+ } else {
+ LOG.error("Received null response when aborting multipart uploads for bucket {}/{}",
+ volumeName, bucketName);
+ }
+ } catch (ServiceException e) {
+ LOG.error("Failed to submit abort multipart uploads request for bucket {}/{}",
+ volumeName, bucketName, e);
+ }
+ }
+
+ private void abortExpiredMultipartUploadsAndClear(OmBucketInfo bucketInfo,
+ PartCountLimitedList expiredUploads) {
+ if (expiredUploads.isEmpty()) {
+ return;
+ }
+
+ abortExpiredMultipartUploads(bucketInfo, expiredUploads.getUploads());
+ expiredUploads.clear();
+ }
+
/**
* If prefix is /dir1/dir2, but dir1 doesn't exist, then it will return exception.
* If prefix is /dir1/dir2, but dir2 doesn't exist, then it will return a list with dir1 only.
@@ -786,6 +1003,7 @@ private void onFailure(String bucketName) {
metrics.incrNumFailureTask();
metrics.incNumKeyIterated(numKeyIterated);
metrics.incNumDirIterated(numDirIterated);
+ metrics.incNumMultipartUploadIterated(numMultipartUploadIterated);
}
private void onSuccess(String bucketName) {
@@ -795,9 +1013,13 @@ private void onSuccess(String bucketName) {
metrics.incTaskLatencyMs(timeSpent);
metrics.incNumKeyIterated(numKeyIterated);
metrics.incNumDirIterated(numDirIterated);
- LOG.info("Spend {} ms on bucket {} to iterate {} keys and {} dirs, deleted {} keys with {} bytes, " +
- "and {} dirs, renamed {} keys with {} bytes, and {} dirs to trash", timeSpent, bucketName, numKeyIterated,
- numDirIterated, numKeyDeleted, sizeKeyDeleted, numDirDeleted, numKeyRenamed, sizeKeyRenamed, numDirRenamed);
+ metrics.incNumMultipartUploadIterated(numMultipartUploadIterated);
+ metrics.incNumMultipartUploadAborted(numMultipartUploadAborted);
+ LOG.info("Spend {} ms on bucket {} to iterate {} keys and {} dirs and {} multipart uploads, " +
+ "deleted {} keys with {} bytes, and {} dirs, renamed {} keys with {} bytes, and {} dirs to trash, " +
+ "aborted {} multipart uploads", timeSpent, bucketName, numKeyIterated,
+ numDirIterated, numMultipartUploadIterated, numKeyDeleted, sizeKeyDeleted, numDirDeleted,
+ numKeyRenamed, sizeKeyRenamed, numDirRenamed, numMultipartUploadAborted);
}
private void handleAndClearFullList(OmBucketInfo bucket, LimitedExpiredObjectList keysList, boolean dir) {
@@ -1066,6 +1288,11 @@ public void setListMaxSize(int size) {
this.listMaxSize = size;
}
+ @VisibleForTesting
+ public void setMpuAbortLimitPerTask(int limit) {
+ this.mpuAbortLimitPerTask = limit;
+ }
+
@VisibleForTesting
public void setOzoneTrash(OzoneTrash ozoneTrash) {
this.ozoneTrash = ozoneTrash;
@@ -1197,6 +1424,61 @@ public void clear() {
}
}
+ /**
+ * A list that tracks the total part count of multipart uploads.
+ * The list is considered "full" when the total part count reaches the limit.
+ * This is used because some MPUs might have 1 part while others might have 10,000 parts.
+ */
+ public static class PartCountLimitedList {
+ private final List<OmMultipartUpload> uploads;
+ private final int maxPartCount;
+ private int currentPartCount;
+
+ public PartCountLimitedList(int maxPartCount) {
+ this.maxPartCount = maxPartCount;
+ this.uploads = new ArrayList<>();
+ this.currentPartCount = 0;
+ }
+
+ /**
+ * Add a multipart upload with its part count.
+ * Caller should check isFull() before calling this method.
+ */
+ public void add(OmMultipartUpload upload, int partCount) {
+ uploads.add(upload);
+ currentPartCount += partCount;
+ }
+
+ public List<OmMultipartUpload> getUploads() {
+ return uploads;
+ }
+
+ public int size() {
+ return uploads.size();
+ }
+
+ public int getPartCount() {
+ return currentPartCount;
+ }
+
+ public boolean isEmpty() {
+ return uploads.isEmpty();
+ }
+
+ public boolean isFull() {
+ boolean full = currentPartCount >= maxPartCount;
+ if (full) {
+ LOG.debug("PartCountLimitedList has reached maximum part count {}", maxPartCount);
+ }
+ return full;
+ }
+
+ public void clear() {
+ uploads.clear();
+ currentPartCount = 0;
+ }
+ }
+
/**
* An in-memory class to hold the information required in directory recursive evaluation.
*/
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleServiceMetrics.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleServiceMetrics.java
index 54539a75c92..b4ffe055396 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleServiceMetrics.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleServiceMetrics.java
@@ -62,6 +62,11 @@ public final class KeyLifecycleServiceMetrics {
private MutableGaugeLong sizeKeyDeleted;
@Metric("Total size of keys renamed")
private MutableGaugeLong sizeKeyRenamed;
+ @Metric("Number of multipart uploads iterated")
+ private MutableGaugeLong numMultipartUploadsIterated;
+
+ @Metric("Total multipart uploads aborted")
+ private MutableGaugeLong numMultipartUploadsAborted;
private KeyLifecycleServiceMetrics() {
this.registry = new MetricsRegistry(METRICS_SOURCE_NAME);
@@ -168,4 +173,24 @@ public void incNumKeyIterated(long keyCount) {
public void incNumDirIterated(long dirCount) {
numDirIterated.incr(dirCount);
}
+
+ public void incNumMultipartUploadIterated(long count) {
+ numMultipartUploadsIterated.incr(count);
+ }
+
+ public void incNumMultipartUploadAborted(long count) {
+ numMultipartUploadsAborted.incr(count);
+ }
+
+ public MutableGaugeLong getNumMultipartUploadsIterated() {
+ return numMultipartUploadsIterated;
+ }
+
+ public MutableGaugeLong getNumMultipartUploadsAborted() {
+ return numMultipartUploadsAborted;
+ }
+
+ public MutableGaugeLong getNumSuccessTask() {
+ return numSuccessTask;
+ }
}
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 e499e0dd9ef..cd2191633ee 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
@@ -36,6 +36,7 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
@@ -71,6 +72,7 @@
import org.apache.hadoop.hdds.server.ServerUtils;
import org.apache.hadoop.hdds.utils.db.DBConfigFromFile;
import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
import org.apache.hadoop.ozone.OzoneAcl;
@@ -91,11 +93,14 @@
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
+import org.apache.hadoop.ozone.om.helpers.OmLCAbortIncompleteMultipartUpload;
import org.apache.hadoop.ozone.om.helpers.OmLCExpiration;
import org.apache.hadoop.ozone.om.helpers.OmLCFilter;
import org.apache.hadoop.ozone.om.helpers.OmLCRule;
import org.apache.hadoop.ozone.om.helpers.OmLifecycleConfiguration;
import org.apache.hadoop.ozone.om.helpers.OmLifecycleRuleAndOperator;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
import org.apache.hadoop.ozone.om.helpers.OpenKeySession;
import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
@@ -1607,6 +1612,257 @@ void testDisableMoveToTrashDeletesDirectly() throws Exception {
assertEquals(0, getKeyCount(FILE_SYSTEM_OPTIMIZED) - initialKeyCount);
deleteLifecyclePolicy(volumeName, bucketName);
}
+
+ @Test
+ void testAbortIncompleteMultipartUploadWithFilters() throws Exception {
+ final String volumeName = getTestName();
+ final String bucketName = uniqueObjectName("bucket");
+
+ // Create volume and bucket
+ createVolumeAndBucket(volumeName, bucketName, OBJECT_STORE,
+ UserGroupInformation.getCurrentUser().getShortUserName());
+
+ String owner = UserGroupInformation.getCurrentUser().getShortUserName();
+
+ long initialMpuCount = getMultipartUploadCount(volumeName, bucketName);
+
+ // Create multipart uploads with different prefixes
+ OmMultipartInfo mpuInfo1 = createTestMultipartUpload(volumeName, bucketName,
+ "uploads/file1", owner); // should match "uploads/" prefix rule
+ OmMultipartInfo mpuInfo2 = createTestMultipartUpload(volumeName, bucketName,
+ "uploads/file2", owner); // should match "uploads/" prefix rule
+ OmMultipartInfo mpuInfo3 = createTestMultipartUpload(volumeName, bucketName,
+ "temp/file3", owner); // should match "temp/" prefix rule
+ OmMultipartInfo mpuInfo4 = createTestMultipartUpload(volumeName, bucketName,
+ "keep/file4", owner); // should NOT match any rule
+
+ // Update creation time to be 2 days ago for all MPUs so they are eligible for abort
+ long oldCreationTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "uploads/file1",
+ mpuInfo1.getUploadID(), oldCreationTime);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "uploads/file2",
+ mpuInfo2.getUploadID(), oldCreationTime);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "temp/file3",
+ mpuInfo3.getUploadID(), oldCreationTime);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "keep/file4",
+ mpuInfo4.getUploadID(), oldCreationTime);
+
+ List<OmLCRule> rules = new ArrayList<>();
+
+ // Rule 1: Abort MPUs with prefix "uploads/" after 1 day
+ OmLCRule rule1 = new OmLCRule.Builder()
+ .setId("abort-mpu-uploads")
+ .setEnabled(true)
+ .setFilter(new OmLCFilter.Builder()
+ .setPrefix("uploads/")
+ .build())
+ .setAction(new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(1)
+ .build())
+ .build();
+ rules.add(rule1);
+
+ // Rule 2: Abort MPUs with prefix "temp/" after 1 day
+ OmLCRule rule2 = new OmLCRule.Builder()
+ .setId("abort-mpu-temp")
+ .setEnabled(true)
+ .setFilter(new OmLCFilter.Builder()
+ .setPrefix("temp/")
+ .build())
+ .setAction(new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(1)
+ .build())
+ .build();
+ rules.add(rule2);
+
+ // Validate the rules have AbortIncompleteMultipartUpload actions
+ for (OmLCRule rule : rules) {
+ assertNotNull(rule.getAbortIncompleteMultipartUpload(),
+ "Rule should have AbortIncompleteMultipartUpload action");
+ assertEquals(1, rule.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
+ createLifecyclePolicy(volumeName, bucketName, OBJECT_STORE, rules);
+
+ // Verify lifecycle configuration was stored correctly
+ String lcKey = "/" + volumeName + "/" + bucketName;
+ OmLifecycleConfiguration storedConfig = metadataManager.getLifecycleConfigurationTable()
+ .get(lcKey);
+ assertNotNull(storedConfig, "Lifecycle configuration should be stored");
+ assertEquals(2, storedConfig.getRules().size(), "Should have 2 rules");
+
+ // Verify rules preserve AbortIncompleteMultipartUpload actions after serialization/deserialization
+ for (OmLCRule rule : storedConfig.getRules()) {
+ assertNotNull(rule.getAbortIncompleteMultipartUpload(),
+ "Stored rule should have AbortIncompleteMultipartUpload action");
+ assertEquals(1, rule.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
+
+ // Verify the lifecycle configuration is valid
+ storedConfig.valid();
+
+ // Wait for lifecycle service to abort the matching MPUs
+ // 3 MPUs should be aborted (uploads/file1, uploads/file2, temp/file3)
+ // 1 MPU should remain (keep/file4)
+ GenericTestUtils.waitFor(() ->
+ getMultipartUploadCount(volumeName, bucketName) - initialMpuCount == 1,
+ WAIT_CHECK_INTERVAL, 10000);
+
+ // Verify only the non-matching MPU remains
+ String keepMpuKey = metadataManager.getMultipartKey(volumeName, bucketName,
+ "keep/file4", mpuInfo4.getUploadID());
+ assertNotNull(metadataManager.getMultipartInfoTable().get(keepMpuKey),
+ "MPU with prefix 'keep/' should NOT be aborted");
+
+ // Verify matching MPUs are aborted
+ String abortedKey1 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "uploads/file1", mpuInfo1.getUploadID());
+ assertNull(metadataManager.getMultipartInfoTable().get(abortedKey1),
+ "MPU with prefix 'uploads/' should be aborted");
+
+ String abortedKey2 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "uploads/file2", mpuInfo2.getUploadID());
+ assertNull(metadataManager.getMultipartInfoTable().get(abortedKey2),
+ "MPU with prefix 'uploads/' should be aborted");
+
+ String abortedKey3 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "temp/file3", mpuInfo3.getUploadID());
+ assertNull(metadataManager.getMultipartInfoTable().get(abortedKey3),
+ "MPU with prefix 'temp/' should be aborted");
+
+ deleteLifecyclePolicy(volumeName, bucketName);
+ }
+
+ @Test
+ void testAbortIncompleteMultipartUploadWithTagFilter() throws Exception {
+ final String volumeName = getTestName();
+ final String bucketName = uniqueObjectName("bucket");
+
+ // Create volume and bucket
+ createVolumeAndBucket(volumeName, bucketName, OBJECT_STORE,
+ UserGroupInformation.getCurrentUser().getShortUserName());
+
+ String owner = UserGroupInformation.getCurrentUser().getShortUserName();
+
+ // Record initial MPU count for this bucket
+ long initialMpuCount = getMultipartUploadCount(volumeName, bucketName);
+
+ // Create multipart uploads with different tags
+ // MPU 1: has matching tag (environment=test) - should be aborted
+ OmKeyArgs keyArgs1 = new OmKeyArgs.Builder()
+ .setVolumeName(volumeName)
+ .setBucketName(bucketName)
+ .setKeyName("file1.txt")
+ .setAcls(Collections.emptyList())
+ .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+ .setLocationInfoList(new ArrayList<>())
+ .setOwnerName(owner)
+ .addTag("environment", "test")
+ .build();
+ OmMultipartInfo mpuInfo1 = writeClient.initiateMultipartUpload(keyArgs1);
+
+ // MPU 2: has matching tag (environment=test) - should be aborted
+ OmKeyArgs keyArgs2 = new OmKeyArgs.Builder()
+ .setVolumeName(volumeName)
+ .setBucketName(bucketName)
+ .setKeyName("file2.txt")
+ .setAcls(Collections.emptyList())
+ .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+ .setLocationInfoList(new ArrayList<>())
+ .setOwnerName(owner)
+ .addTag("environment", "test")
+ .build();
+ OmMultipartInfo mpuInfo2 = writeClient.initiateMultipartUpload(keyArgs2);
+
+ // MPU 3: has non-matching tag value (environment=prod) - should NOT be aborted
+ OmKeyArgs keyArgs3 = new OmKeyArgs.Builder()
+ .setVolumeName(volumeName)
+ .setBucketName(bucketName)
+ .setKeyName("file3.txt")
+ .setAcls(Collections.emptyList())
+ .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+ .setLocationInfoList(new ArrayList<>())
+ .setOwnerName(owner)
+ .addTag("environment", "prod")
+ .build();
+ OmMultipartInfo mpuInfo3 = writeClient.initiateMultipartUpload(keyArgs3);
+
+ // MPU 4: has no tags - should NOT be aborted
+ OmKeyArgs keyArgs4 = new OmKeyArgs.Builder()
+ .setVolumeName(volumeName)
+ .setBucketName(bucketName)
+ .setKeyName("file4.txt")
+ .setAcls(Collections.emptyList())
+ .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+ .setLocationInfoList(new ArrayList<>())
+ .setOwnerName(owner)
+ .build();
+ OmMultipartInfo mpuInfo4 = writeClient.initiateMultipartUpload(keyArgs4);
+
+ // Update creation time to be 2 days ago for all MPUs so they are eligible for abort
+ long oldCreationTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(2);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "file1.txt",
+ mpuInfo1.getUploadID(), oldCreationTime);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "file2.txt",
+ mpuInfo2.getUploadID(), oldCreationTime);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "file3.txt",
+ mpuInfo3.getUploadID(), oldCreationTime);
+ updateMultipartUploadCreationTime(volumeName, bucketName, "file4.txt",
+ mpuInfo4.getUploadID(), oldCreationTime);
+
+ List<OmLCRule> rules = new ArrayList<>();
+
+ // Rule: Abort MPUs with tag environment=test after 1 day
+ OmLCRule rule = new OmLCRule.Builder()
+ .setId("abort-mpu-test-env")
+ .setEnabled(true)
+ .setFilter(new OmLCFilter.Builder()
+ .setTag("environment", "test")
+ .build())
+ .setAction(new OmLCAbortIncompleteMultipartUpload.Builder()
+ .setDaysAfterInitiation(1)
+ .build())
+ .build();
+ rules.add(rule);
+
+ // Validate the rule has AbortIncompleteMultipartUpload action
+ assertNotNull(rule.getAbortIncompleteMultipartUpload(),
+ "Rule should have AbortIncompleteMultipartUpload action");
+
+ createLifecyclePolicy(volumeName, bucketName, OBJECT_STORE, rules);
+
+ // Wait for lifecycle service to abort the matching MPUs
+ // 2 MPUs should be aborted (file1.txt and file2.txt with environment=test)
+ // 2 MPUs should remain (file3.txt with environment=prod and file4.txt with no tags)
+ GenericTestUtils.waitFor(() ->
+ getMultipartUploadCount(volumeName, bucketName) - initialMpuCount == 2,
+ WAIT_CHECK_INTERVAL, 10000);
+
+ // Verify non-matching MPUs remain
+ String remainKey3 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "file3.txt", mpuInfo3.getUploadID());
+ assertNotNull(metadataManager.getMultipartInfoTable().get(remainKey3),
+ "MPU with tag 'environment=prod' should NOT be aborted");
+
+ String remainKey4 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "file4.txt", mpuInfo4.getUploadID());
+ assertNotNull(metadataManager.getMultipartInfoTable().get(remainKey4),
+ "MPU without tags should NOT be aborted");
+
+ // Verify matching MPUs are aborted
+ String abortedKey1 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "file1.txt", mpuInfo1.getUploadID());
+ assertNull(metadataManager.getMultipartInfoTable().get(abortedKey1),
+ "MPU with tag 'environment=test' should be aborted");
+
+ String abortedKey2 = metadataManager.getMultipartKey(volumeName, bucketName,
+ "file2.txt", mpuInfo2.getUploadID());
+ assertNull(metadataManager.getMultipartInfoTable().get(abortedKey2),
+ "MPU with tag 'environment=test' should be aborted");
+
+ deleteLifecyclePolicy(volumeName, bucketName);
+ }
+
}
/**
@@ -2055,6 +2311,68 @@ private long getKeyCount(BucketLayout layout) {
}
}
+ private long getMultipartUploadCount(String volumeName, String bucketName) {
+ String prefix = metadataManager.getBucketKeyPrefix(volumeName, bucketName);
+ long count = 0;
+ try (TableIterator<String, ? extends Table.KeyValue<String, OmMultipartKeyInfo>> iter =
+ metadataManager.getMultipartInfoTable().iterator(prefix)) {
+ while (iter.hasNext()) {
+ Table.KeyValue<String, OmMultipartKeyInfo> entry = iter.next();
+ // Check if the key starts with the prefix (belongs to this bucket)
+ if (entry.getKey().startsWith(prefix)) {
+ count++;
+ } else {
+ break; // Iterator went past our prefix
+ }
+ }
+ } catch (IOException e) {
+ fail("Failed to count multipart uploads: " + e.getMessage());
+ return -1;
+ }
+ return count;
+ }
+
+ private void updateMultipartUploadCreationTime(String volumeName, String bucketName,
+ String keyName, String uploadId, long newCreationTime) throws IOException {
+ String dbKey = metadataManager.getMultipartKey(volumeName, bucketName, keyName, uploadId);
+ OmMultipartKeyInfo existingInfo = metadataManager.getMultipartInfoTable().get(dbKey);
+ if (existingInfo == null) {
+ fail("Multipart upload not found: " + dbKey);
+ return;
+ }
+
+ // Create a new OmMultipartKeyInfo with the updated creation time
+ OmMultipartKeyInfo updatedInfo = new OmMultipartKeyInfo.Builder()
+ .setUploadID(existingInfo.getUploadID())
+ .setCreationTime(newCreationTime)
+ .setReplicationConfig(existingInfo.getReplicationConfig())
+ .setObjectID(existingInfo.getObjectID())
+ .setUpdateID(existingInfo.getUpdateID())
+ .setParentID(existingInfo.getParentID())
+ .build();
+
+ // Copy part key infos
+ for (OzoneManagerProtocolProtos.PartKeyInfo partKeyInfo : existingInfo.getPartKeyInfoMap()) {
+ updatedInfo.addPartKeyInfo(partKeyInfo);
+ }
+
+ metadataManager.getMultipartInfoTable().put(dbKey, updatedInfo);
+ }
+
+ private OmMultipartInfo createTestMultipartUpload(String volumeName, String bucketName,
+ String keyName, String owner) throws IOException {
+ OmKeyArgs keyArgs = new OmKeyArgs.Builder()
+ .setVolumeName(volumeName)
+ .setBucketName(bucketName)
+ .setKeyName(keyName)
+ .setAcls(Collections.emptyList())
+ .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+ .setLocationInfoList(new ArrayList<>())
+ .setOwnerName(owner)
+ .build();
+ return writeClient.initiateMultipartUpload(keyArgs);
+ }
+
public static String uniqueObjectName(String prefix) {
return prefix + OBJECT_COUNTER.getAndIncrement();
}
diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/S3LifecycleConfiguration.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/S3LifecycleConfiguration.java
index 5cbc4a275d2..ff72a322dee 100644
--- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/S3LifecycleConfiguration.java
+++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/S3LifecycleConfiguration.java
@@ -28,6 +28,7 @@
import org.apache.hadoop.ozone.client.OzoneBucket;
import org.apache.hadoop.ozone.client.OzoneLifecycleConfiguration;
import org.apache.hadoop.ozone.om.exceptions.OMException;
+import org.apache.hadoop.ozone.om.helpers.OmLCAbortIncompleteMultipartUpload;
import org.apache.hadoop.ozone.om.helpers.OmLCExpiration;
import org.apache.hadoop.ozone.om.helpers.OmLCFilter;
import org.apache.hadoop.ozone.om.helpers.OmLCRule;
@@ -72,6 +73,9 @@ public static class Rule {
@XmlElement(name = "Expiration")
private Expiration expiration;
+ @XmlElement(name = "AbortIncompleteMultipartUpload")
+ private AbortIncompleteMultipartUpload abortIncompleteMultipartUpload;
+
@XmlElement(name = "Filter")
private Filter filter;
@@ -107,6 +111,14 @@ public void setExpiration(Expiration expiration) {
this.expiration = expiration;
}
+ public AbortIncompleteMultipartUpload getAbortIncompleteMultipartUpload() {
+ return abortIncompleteMultipartUpload;
+ }
+
+ public void setAbortIncompleteMultipartUpload(AbortIncompleteMultipartUpload abortIncompleteMultipartUpload) {
+ this.abortIncompleteMultipartUpload = abortIncompleteMultipartUpload;
+ }
+
public Filter getFilter() {
return filter;
}
@@ -145,6 +157,24 @@ public void setDate(String date) {
}
}
+ /**
+ * AbortIncompleteMultipartUpload entity for lifecycle rule.
+ */
+ @XmlAccessorType(XmlAccessType.FIELD)
+ @XmlRootElement(name = "AbortIncompleteMultipartUpload")
+ public static class AbortIncompleteMultipartUpload {
+ @XmlElement(name = "DaysAfterInitiation")
+ private Integer daysAfterInitiation;
+
+ public Integer getDaysAfterInitiation() {
+ return daysAfterInitiation;
+ }
+
+ public void setDaysAfterInitiation(Integer daysAfterInitiation) {
+ this.daysAfterInitiation = daysAfterInitiation;
+ }
+ }
+
/**
* Tag entity for filter criteria.
*/
@@ -293,7 +323,10 @@ private OmLCRule convertToOmRule(Rule rule) throws OMException, OS3Exception {
.setPrefix(rule.getPrefix());
if (rule.getExpiration() != null) {
- builder.setAction(convertToOmExpiration(rule.getExpiration()));
+ builder.addAction(convertToOmExpiration(rule.getExpiration()));
+ }
+ if (rule.getAbortIncompleteMultipartUpload() != null) {
+ builder.addAction(convertToOmAbortIncompleteMultipartUpload(rule.getAbortIncompleteMultipartUpload()));
}
if (rule.getFilter() != null) {
builder.setFilter(convertToOmFilter(rule.getFilter()));
@@ -321,6 +354,24 @@ private OmLCExpiration convertToOmExpiration(Expiration expiration) throws OMExc
return builder.build();
}
+ /**
+ * Converts S3 AbortIncompleteMultipartUpload to internal representation.
+ *
+ * @param abortIncompleteMultipartUpload the S3 AbortIncompleteMultipartUpload
+ * @return OmLCAbortIncompleteMultipartUpload internal representation
+ */
+ private OmLCAbortIncompleteMultipartUpload convertToOmAbortIncompleteMultipartUpload(
+ AbortIncompleteMultipartUpload abortIncompleteMultipartUpload) throws OMException {
+ OmLCAbortIncompleteMultipartUpload.Builder builder =
+ new OmLCAbortIncompleteMultipartUpload.Builder();
+
+ if (abortIncompleteMultipartUpload.getDaysAfterInitiation() != null) {
+ builder.setDaysAfterInitiation(abortIncompleteMultipartUpload.getDaysAfterInitiation());
+ }
+
+ return builder.build();
+ }
+
/**
* Converts S3 filter to internal filter.
*
@@ -402,6 +453,10 @@ private static Rule convertFromOzoneRule(OzoneLifecycleConfiguration.OzoneLCRule
if (ozoneRule.getExpiration() != null) {
rule.setExpiration(convertFromOzoneExpiration(ozoneRule.getExpiration()));
}
+ if (ozoneRule.getAbortIncompleteMultipartUpload() != null) {
+ rule.setAbortIncompleteMultipartUpload(
+ convertFromOzoneAbortIncompleteMultipartUpload(ozoneRule.getAbortIncompleteMultipartUpload()));
+ }
if (ozoneRule.getFilter() != null) {
rule.setFilter(convertFromOzoneFilter(ozoneRule.getFilter()));
}
@@ -431,6 +486,25 @@ private static Expiration convertFromOzoneExpiration(
return expiration;
}
+ /**
+ * Converts an Ozone internal AbortIncompleteMultipartUpload to S3 representation.
+ *
+ * @param ozoneAbortIncompleteMultipartUpload internal AbortIncompleteMultipartUpload
+ * @return AbortIncompleteMultipartUpload S3 representation
+ */
+ private static AbortIncompleteMultipartUpload convertFromOzoneAbortIncompleteMultipartUpload(
+ OzoneLifecycleConfiguration.OzoneLCAbortIncompleteMultipartUpload ozoneAbortIncompleteMultipartUpload) {
+
+ AbortIncompleteMultipartUpload abortIncompleteMultipartUpload = new AbortIncompleteMultipartUpload();
+
+ if (ozoneAbortIncompleteMultipartUpload.getDaysAfterInitiation() > 0) {
+ abortIncompleteMultipartUpload.setDaysAfterInitiation(
+ ozoneAbortIncompleteMultipartUpload.getDaysAfterInitiation());
+ }
+
+ return abortIncompleteMultipartUpload;
+ }
+
/**
* Converts an Ozone internal filter to S3 filter.
*
diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/OzoneBucketStub.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/OzoneBucketStub.java
index 5a823373e45..fbfb72afed0 100644
--- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/OzoneBucketStub.java
+++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/OzoneBucketStub.java
@@ -801,12 +801,17 @@ private static OzoneLifecycleConfiguration toOzoneLifecycleConfiguration(
for (OmLCRule r: omLifecycleConfiguration.getRules()) {
OzoneLifecycleConfiguration.OzoneLCExpiration e = null;
+ OzoneLifecycleConfiguration.OzoneLCAbortIncompleteMultipartUpload a = null;
OzoneLifecycleConfiguration.OzoneLCFilter f = null;
if (r.getExpiration() != null) {
e = new OzoneLifecycleConfiguration.OzoneLCExpiration(
r.getExpiration().getDays(), r.getExpiration().getDate());
}
+ if (r.getAbortIncompleteMultipartUpload() != null) {
+ a = new OzoneLifecycleConfiguration.OzoneLCAbortIncompleteMultipartUpload(
+ r.getAbortIncompleteMultipartUpload().getDaysAfterInitiation());
+ }
if (r.getFilter() != null) {
OzoneLifecycleConfiguration.LifecycleAndOperator andOperator = null;
if (r.getFilter().getAndOperator() != null) {
@@ -818,7 +823,7 @@ private static OzoneLifecycleConfiguration toOzoneLifecycleConfiguration(
}
rules.add(new OzoneLifecycleConfiguration.OzoneLCRule(r.getId(),
- r.getEffectivePrefix(), (r.isEnabled() ? "Enabled" : "Disabled"), e, f));
+ r.getEffectivePrefix(), (r.isEnabled() ? "Enabled" : "Disabled"), e, a, f));
}
return new OzoneLifecycleConfiguration(
diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationGet.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationGet.java
index 2d602b4f4fc..0558af9c1f8 100644
--- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationGet.java
+++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationGet.java
@@ -83,6 +83,47 @@ public void testGetLifecycleConfiguration() throws Exception {
lcc.getRules().get(0).getExpiration().getDays().intValue());
}
+ @Test
+ public void testGetLifecycleWithAbortIncompleteMultipartUpload() throws Exception {
+ String bucketName = "bucket1";
+ bucketEndpoint.put(bucketName, getBodyWithAbortAction());
+ Response r = bucketEndpoint.get(bucketName);
+
+ assertEquals(HTTP_OK, r.getStatus());
+ S3LifecycleConfiguration lcc = (S3LifecycleConfiguration) r.getEntity();
+ assertEquals(1, lcc.getRules().size());
+ S3LifecycleConfiguration.Rule rule = lcc.getRules().get(0);
+
+ assertEquals("abort-incomplete-uploads", rule.getId());
+ assertEquals("uploads/", rule.getPrefix());
+ assertEquals("Enabled", rule.getStatus());
+ assertEquals(7, rule.getAbortIncompleteMultipartUpload()
+ .getDaysAfterInitiation().intValue());
+ }
+
+ @Test
+ public void testGetLifecycleWithBothActions() throws Exception {
+ String bucketName = "bucket1";
+ bucketEndpoint.put(bucketName, getBodyWithBothActions());
+ Response r = bucketEndpoint.get(bucketName);
+
+ assertEquals(HTTP_OK, r.getStatus());
+ S3LifecycleConfiguration lcc = (S3LifecycleConfiguration) r.getEntity();
+ assertEquals(1, lcc.getRules().size());
+ S3LifecycleConfiguration.Rule rule = lcc.getRules().get(0);
+
+ assertEquals("cleanup-rule", rule.getId());
+ assertEquals("temp/", rule.getPrefix());
+ assertEquals("Enabled", rule.getStatus());
+
+ // Verify Expiration action
+ assertEquals(30, rule.getExpiration().getDays().intValue());
+
+ // Verify AbortIncompleteMultipartUpload action
+ assertEquals(7, rule.getAbortIncompleteMultipartUpload()
+ .getDaysAfterInitiation().intValue());
+ }
+
private static InputStream getBody() {
String xml = ("<LifecycleConfiguration xmlns=\"http://s3.amazonaws" +
".com/doc/2006-03-01/\">" +
@@ -96,4 +137,37 @@ private static InputStream getBody() {
return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
}
+
+ private static InputStream getBodyWithAbortAction() {
+ String xml = "<LifecycleConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" +
+ "<Rule>" +
+ "<ID>abort-incomplete-uploads</ID>" +
+ "<Prefix>uploads/</Prefix>" +
+ "<Status>Enabled</Status>" +
+ "<AbortIncompleteMultipartUpload>" +
+ "<DaysAfterInitiation>7</DaysAfterInitiation>" +
+ "</AbortIncompleteMultipartUpload>" +
+ "</Rule>" +
+ "</LifecycleConfiguration>";
+
+ return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static InputStream getBodyWithBothActions() {
+ String xml = "<LifecycleConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" +
+ "<Rule>" +
+ "<ID>cleanup-rule</ID>" +
+ "<Prefix>temp/</Prefix>" +
+ "<Status>Enabled</Status>" +
+ "<Expiration>" +
+ "<Days>30</Days>" +
+ "</Expiration>" +
+ "<AbortIncompleteMultipartUpload>" +
+ "<DaysAfterInitiation>7</DaysAfterInitiation>" +
+ "</AbortIncompleteMultipartUpload>" +
+ "</Rule>" +
+ "</LifecycleConfiguration>";
+
+ return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
+ }
}
diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationPut.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationPut.java
index f0041e12ec8..d6274b60f60 100644
--- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationPut.java
+++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestS3LifecycleConfigurationPut.java
@@ -181,6 +181,29 @@ public void testPutLifecycleConfigurationFailsWithNonBucketOwner()
}
}
+ @Test
+ public void testPutLifecycleWithAbortIncompleteMultipartUpload() throws Exception {
+ assertEquals(HTTP_OK, bucketEndpoint.put("bucket1", withAbortIncompleteMultipartUpload()).getStatus());
+ }
+
+ @Test
+ public void testPutLifecycleWithBothExpirationAndAbort() throws Exception {
+ assertEquals(HTTP_OK, bucketEndpoint.put("bucket1", withBothExpirationAndAbort()).getStatus());
+ }
+
+ @Test
+ public void testPutInvalidAbortIncompleteMultipartUploadConfig() throws Exception {
+ // Test with zero days - should fail
+ testInvalidLifecycleConfiguration(
+ TestS3LifecycleConfigurationPut::withAbortZeroDays,
+ HTTP_BAD_REQUEST, INVALID_REQUEST.getCode());
+
+ // Test with negative days - should fail
+ testInvalidLifecycleConfiguration(
+ TestS3LifecycleConfigurationPut::withAbortNegativeDays,
+ HTTP_BAD_REQUEST, INVALID_REQUEST.getCode());
+ }
+
private static InputStream onePrefix() {
String xml = ("<LifecycleConfiguration xmlns=\"http://s3.amazonaws" +
".com/doc/2006-03-01/\">" +
@@ -516,4 +539,67 @@ private InputStream useDuplicateTagInAndOperator() {
return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
}
+
+ private static InputStream withAbortIncompleteMultipartUpload() {
+ String xml = "<LifecycleConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" +
+ "<Rule>" +
+ "<ID>abort-incomplete-uploads</ID>" +
+ "<Prefix>uploads/</Prefix>" +
+ "<Status>Enabled</Status>" +
+ "<AbortIncompleteMultipartUpload>" +
+ "<DaysAfterInitiation>7</DaysAfterInitiation>" +
+ "</AbortIncompleteMultipartUpload>" +
+ "</Rule>" +
+ "</LifecycleConfiguration>";
+
+ return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static InputStream withBothExpirationAndAbort() {
+ String xml = "<LifecycleConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" +
+ "<Rule>" +
+ "<ID>cleanup-rule</ID>" +
+ "<Prefix>temp/</Prefix>" +
+ "<Status>Enabled</Status>" +
+ "<Expiration>" +
+ "<Days>30</Days>" +
+ "</Expiration>" +
+ "<AbortIncompleteMultipartUpload>" +
+ "<DaysAfterInitiation>7</DaysAfterInitiation>" +
+ "</AbortIncompleteMultipartUpload>" +
+ "</Rule>" +
+ "</LifecycleConfiguration>";
+
+ return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static InputStream withAbortZeroDays() {
+ String xml = "<LifecycleConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" +
+ "<Rule>" +
+ "<ID>invalid-abort</ID>" +
+ "<Prefix>uploads/</Prefix>" +
+ "<Status>Enabled</Status>" +
+ "<AbortIncompleteMultipartUpload>" +
+ "<DaysAfterInitiation>0</DaysAfterInitiation>" +
+ "</AbortIncompleteMultipartUpload>" +
+ "</Rule>" +
+ "</LifecycleConfiguration>";
+
+ return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private static InputStream withAbortNegativeDays() {
+ String xml = "<LifecycleConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" +
+ "<Rule>" +
+ "<ID>invalid-abort</ID>" +
+ "<Prefix>uploads/</Prefix>" +
+ "<Status>Enabled</Status>" +
+ "<AbortIncompleteMultipartUpload>" +
+ "<DaysAfterInitiation>-1</DaysAfterInitiation>" +
+ "</AbortIncompleteMultipartUpload>" +
+ "</Rule>" +
+ "</LifecycleConfiguration>";
+
+ return new ByteArrayInputStream(xml.getBytes(StandardCharsets.UTF_8));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org