This is an automated email from the ASF dual-hosted git repository.
chungen0126 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 24e06a40c16 HDDS-15835. Count incomplete MPU parts in quota repair
(#10795)
24e06a40c16 is described below
commit 24e06a40c16df03757031acfdc94ddf01642890d
Author: Chi-Hsuan Huang <[email protected]>
AuthorDate: Wed Sep 30 16:20:24 2026 +0800
HDDS-15835. Count incomplete MPU parts in quota repair (#10795)
---
.../hadoop/ozone/om/service/QuotaRepairTask.java | 64 ++++-
.../ozone/om/service/TestQuotaRepairTask.java | 306 +++++++++++++++++++++
2 files changed, 367 insertions(+), 3 deletions(-)
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java
index 45e41e4dc34..bc800b5a5b2 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java
@@ -38,6 +38,7 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.SortedMap;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
@@ -68,9 +69,13 @@
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartInfo;
+import org.apache.hadoop.ozone.om.helpers.QuotaUtil;
import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils;
+import org.apache.hadoop.ozone.om.request.util.OMMultipartUploadUtils;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.util.Time;
import org.apache.ratis.protocol.ClientId;
@@ -89,9 +94,9 @@ public class QuotaRepairTask {
private static final int TASK_THREAD_CNT = 3;
/**
* Parallel full-table scans: OBS keys, FSO files, dirs, active deleted
keys/dirs,
- * snapshot DB deleted keys/dirs.
+ * snapshot DB deleted keys/dirs, multipart upload parts.
*/
- private static final int QUOTA_REPAIR_SCAN_TASKS = 6;
+ private static final int QUOTA_REPAIR_SCAN_TASKS = 7;
private static final AtomicBoolean IN_PROGRESS = new AtomicBoolean(false);
private static final RepairStatus REPAIR_STATUS = new RepairStatus();
private static final AtomicLong RUN_CNT = new AtomicLong(0);
@@ -332,6 +337,7 @@ private void repairCount(
Map<String, CountPair> directoryCountMap = new ConcurrentHashMap<>();
Map<String, CountPair> snapshotDeletedKeyMap = new ConcurrentHashMap<>();
Map<String, CountPair> snapshotDeletedDirMap = new ConcurrentHashMap<>();
+ Map<String, CountPair> mpuCountMap = new ConcurrentHashMap<>();
try {
nameBucketInfoMap.keySet().stream().forEach(e -> keyCountMap.put(e,
new CountPair()));
@@ -341,6 +347,7 @@ private void repairCount(
new CountPair()));
nameBucketInfoMap.keySet().forEach(k -> snapshotDeletedKeyMap.put(k, new
CountPair()));
idBucketInfoMap.keySet().forEach(k -> snapshotDeletedDirMap.put(k, new
CountPair()));
+ nameBucketInfoMap.keySet().forEach(k -> mpuCountMap.put(k, new
CountPair()));
List<Future<?>> tasks = new ArrayList<>();
tasks.add(executor.submit(() -> recalculateUsages(
@@ -369,6 +376,7 @@ private void repairCount(
throw new UncheckedIOException(ex);
}
}));
+ tasks.add(executor.submit(() ->
recalculateMultipartUsages(metadataManager, mpuCountMap)));
// await every scan before propagating a failure, so no scan outlives
the checkpoint
Exception scanFailure = awaitAll(tasks, null);
@@ -387,6 +395,7 @@ private void repairCount(
updateCountToBucketInfo(nameBucketInfoMap, keyCountMap);
updateCountToBucketInfo(idBucketInfoMap, fileCountMap);
updateCountToBucketInfo(idBucketInfoMap, directoryCountMap);
+ updateCountToBucketInfo(nameBucketInfoMap, mpuCountMap);
mergeSnapshotDeletedTableCounts(nameBucketInfoMap, snapshotDeletedKeyMap);
mergeDeletedDirSnapshotNamespace(idBucketInfoMap, snapshotDeletedDirMap);
LOG.info("Completed quota repair counting for all keys, files and
directories");
@@ -558,6 +567,54 @@ private void recalculateDeletedDirNamespace(
}
}
+ /**
+ * Counts committed parts of incomplete multipart uploads from the active DB
checkpoint.
+ * Parts of a snapshot's incomplete uploads stay charged to that snapshot's
own bucket,
+ * so only the active multipartInfoTable is scanned here. The scan reads
keys only and
+ * each counting task reads the row it needs, so a row is never decoded for
a bucket
+ * outside the repair and the batches hold keys instead of parts.
+ */
+ private void recalculateMultipartUsages(
+ OMMetadataManager metadataManager, Map<String, CountPair> mpuCountMap)
+ throws UncheckedIOException, UncheckedExecutionException {
+ try (Table.KeyValueIterator<String, OmMultipartKeyInfo> keyIter
+ = metadataManager.getMultipartInfoTable().iterator(null, KEY_ONLY)) {
+ scanTableInBatches(executor, keyIter, "Multipart upload usages",
+ kv -> extractMultipartCount(kv.getKey(), mpuCountMap,
metadataManager));
+ } catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ }
+
+ @VisibleForTesting
+ static void extractMultipartCount(
+ String key, Map<String, CountPair> mpuCountMap,
+ OMMetadataManager metadataManager) throws UncheckedIOException {
+ try {
+ CountPair usage = mpuCountMap.get(getVolumeBucketPrefix(key));
+ if (usage == null) {
+ return;
+ }
+ OmMultipartKeyInfo multipartKeyInfo =
+ metadataManager.getMultipartInfoTable().getSkipCache(key);
+ long replicatedSize = 0;
+ if (multipartKeyInfo.getSchemaVersion() ==
OmMultipartKeyInfo.LEGACY_SCHEMA_VERSION) {
+ for (OzoneManagerProtocolProtos.PartKeyInfo partKeyInfo :
multipartKeyInfo.getPartKeyInfoMap()) {
+ replicatedSize += QuotaUtil.getReplicatedSize(
+ partKeyInfo.getPartKeyInfo().getDataSize(),
multipartKeyInfo.getReplicationConfig());
+ }
+ } else {
+ SortedMap<Integer, OmMultipartPartInfo> parts =
+ OMMultipartUploadUtils.scanParts(metadataManager,
multipartKeyInfo.getUploadID());
+ replicatedSize = OMMultipartUploadUtils.getReplicatedSize(
+ parts, multipartKeyInfo.getReplicationConfig());
+ }
+ usage.incrSpace(replicatedSize);
+ } catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ }
+
private static synchronized void mergeSnapshotDeletedTableCounts(
Map<String, OmBucketInfo> nameBucketInfoMap,
Map<String, CountPair> counts) {
@@ -763,7 +820,8 @@ private static String getVolumeBucketPrefix(String key) {
return prefix;
}
- private static class CountPair {
+ @VisibleForTesting
+ static class CountPair {
private AtomicLong space = new AtomicLong();
private AtomicLong namespace = new AtomicLong();
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java
index 00b01b18c44..3565484b4c5 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestQuotaRepairTask.java
@@ -29,11 +29,17 @@
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
import com.google.common.util.concurrent.UncheckedExecutionException;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -41,23 +47,36 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hadoop.hdds.client.ECReplicationConfig;
import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.utils.db.BatchOperation;
import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.hdds.utils.db.cache.CacheKey;
import org.apache.hadoop.hdds.utils.db.cache.CacheValue;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartKey;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartUpload;
import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
import org.apache.hadoop.ozone.om.ratis.OzoneManagerRatisServer;
import org.apache.hadoop.ozone.om.request.OMRequestTestUtils;
import org.apache.hadoop.ozone.om.request.key.OMKeyRequestTests;
+import
org.apache.hadoop.ozone.om.request.s3.multipart.S3MultipartUploadAbortRequest;
+import
org.apache.hadoop.ozone.om.request.s3.multipart.S3MultipartUploadAbortRequestWithFSO;
import org.apache.hadoop.ozone.om.request.volume.OMQuotaRepairRequest;
import org.apache.hadoop.ozone.om.response.OMClientResponse;
import org.apache.hadoop.ozone.om.response.volume.OMQuotaRepairResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyInfo;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartKeyInfo;
import org.apache.hadoop.util.Time;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -70,6 +89,7 @@ public class TestQuotaRepairTask extends OMKeyRequestTests {
/** Seconds; must match {@link Timeout} on this class. */
private static final int REPAIR_TEST_TIMEOUT_SECONDS = 120;
+ private static final ReplicationConfig RATIS_ONE =
RatisReplicationConfig.getInstance(ONE);
private static Boolean awaitRepair(CompletableFuture<Boolean> repair) throws
Exception {
return repair.get(REPAIR_TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
@@ -449,4 +469,290 @@ public void testInterruptedScanStillAwaitsWorkers()
throws Exception {
executor.shutdownNow();
}
}
+
+ @Test
+ public void testQuotaRepairCountsMPUParts() throws Exception {
+ AtomicReference<OzoneManagerProtocolProtos.OMRequest> request =
mockQuotaRepairRequest();
+ OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName,
+ omMetadataManager, BucketLayout.OBJECT_STORE);
+ String otherBucketName = "other" + bucketName;
+ OMRequestTestUtils.addBucketToDB(volumeName, otherBucketName,
+ omMetadataManager, BucketLayout.OBJECT_STORE);
+
+ // committed keys stay charged after the MPU parts are released
+ int keyCount = 3;
+ for (int i = 0; i < keyCount; i++) {
+ OMRequestTestUtils.addKeyToTableAndCache(volumeName, bucketName, "key" +
i, -1,
+ RatisReplicationConfig.getInstance(ONE), 10L + i, omMetadataManager);
+ }
+ long keyBytes = keyCount * 1000L;
+
+ // legacy schema: parts are inlined in the upload record, keyed by part
number,
+ // so committing part 1 twice (100 then 300) leaves only the current 300
+ String legacyKey = "legacyMpuKey";
+ String legacyUploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo legacyInfo = newMultipartInfo(legacyUploadId, 1001L,
RATIS_ONE, false);
+ legacyInfo.addPartKeyInfo(createPart(bucketName, legacyKey,
legacyUploadId, 1, 100L));
+ legacyInfo.addPartKeyInfo(createPart(bucketName, legacyKey,
legacyUploadId, 1, 300L));
+ legacyInfo.addPartKeyInfo(createPart(bucketName, legacyKey,
legacyUploadId, 2, 200L));
+ addMultipartInfo(bucketName, legacyKey, legacyInfo, 1L);
+ long legacyBytes = 300L + 200L;
+
+ // split-parts schema: parts live in their own table
+ String splitKey = "splitMpuKey";
+ String splitUploadId = UUID.randomUUID().toString();
+ addMultipartInfo(bucketName, splitKey, newMultipartInfo(splitUploadId,
1002L, RATIS_ONE, true), 2L);
+ putSplitPart(bucketName, splitKey, splitUploadId, 1, 400L);
+ putSplitPart(bucketName, splitKey, splitUploadId, 2, 500L);
+ long splitBytes = 400L + 500L;
+
+ // uploads initiated but with no part committed yet must contribute nothing
+ String emptyLegacyUploadId = UUID.randomUUID().toString();
+ addMultipartInfo(bucketName, "emptyLegacyMpuKey",
+ newMultipartInfo(emptyLegacyUploadId, 1003L, RATIS_ONE, false), 3L);
+ String emptySplitUploadId = UUID.randomUUID().toString();
+ addMultipartInfo(bucketName, "emptySplitMpuKey",
+ newMultipartInfo(emptySplitUploadId, 1004L, RATIS_ONE, true), 4L);
+
+ // a second bucket proves parts are charged to the bucket that owns them.
+ // Its upload replicates three ways, so the part size is charged three
times over.
+ String otherKey = "otherMpuKey";
+ String otherUploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo otherInfo = newMultipartInfo(otherUploadId, 2001L,
+ RatisReplicationConfig.getInstance(THREE), false);
+ otherInfo.addPartKeyInfo(createPart(otherBucketName, otherKey,
otherUploadId, 1, 700L));
+ addMultipartInfo(otherBucketName, otherKey, otherInfo, 5L);
+ long otherBytes = 700L * 3;
+
+ String bucketKey = corruptBucketUsage(bucketName, 12345L, 99L, 6L);
+ String otherBucketKey = corruptBucketUsage(otherBucketName, 54321L, 77L,
7L);
+ applyQuotaRepair(request, 8L);
+
+ OmBucketInfo repaired = omMetadataManager.getBucketTable().get(bucketKey);
+ assertEquals(keyBytes + legacyBytes + splitBytes, repaired.getUsedBytes());
+ // MPU parts consume no namespace, only the committed keys do
+ assertEquals(keyCount, repaired.getUsedNamespace());
+ OmBucketInfo otherRepaired =
omMetadataManager.getBucketTable().get(otherBucketKey);
+ assertEquals(otherBytes, otherRepaired.getUsedBytes());
+ assertEquals(0L, otherRepaired.getUsedNamespace());
+
+ // aborting through the real request path must release exactly what repair
counted.
+ // No open key is staged, so abort takes its orphan-parts path.
+ abortMpu(bucketName, legacyKey, legacyUploadId, BucketLayout.OBJECT_STORE,
5001L);
+ abortMpu(bucketName, splitKey, splitUploadId, BucketLayout.OBJECT_STORE,
5002L);
+ assertEquals(keyBytes,
omMetadataManager.getBucketTable().get(bucketKey).getUsedBytes());
+ abortMpu(otherBucketName, otherKey, otherUploadId,
BucketLayout.OBJECT_STORE, 5003L);
+ assertEquals(0L,
omMetadataManager.getBucketTable().get(otherBucketKey).getUsedBytes());
+ }
+
+ @Test
+ public void testQuotaRepairCountsMPUPartsWithFSO() throws Exception {
+ AtomicReference<OzoneManagerProtocolProtos.OMRequest> request =
mockQuotaRepairRequest();
+ String fsoBucketName = "fsomup" + bucketName;
+ OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, fsoBucketName,
+ omMetadataManager, BucketLayout.FILE_SYSTEM_OPTIMIZED);
+
+ String keyName = "fsoMpuFile";
+ String uploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo mpuInfo = newMultipartInfo(uploadId, 3001L,
+ RatisReplicationConfig.getInstance(THREE), true);
+ addMultipartInfo(fsoBucketName, keyName, mpuInfo, 1L);
+ putSplitPart(fsoBucketName, keyName, uploadId, 1, 100L);
+ long ratisBytes = 100L * 3;
+
+ // EC parity overhead is charged too: one full rs-3-2-1024k stripe of data
+ // costs the 3 MiB of data plus 2 MiB of parity
+ String ecKeyName = "ecMpuFile";
+ String ecUploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo ecInfo = newMultipartInfo(ecUploadId, 3002L,
+ new ECReplicationConfig("rs-3-2-1024k"), true);
+ addMultipartInfo(fsoBucketName, ecKeyName, ecInfo, 2L);
+ putSplitPart(fsoBucketName, ecKeyName, ecUploadId, 1, 3 * 1024 * 1024L);
+ long ecBytes = 5 * 1024 * 1024L;
+
+ String bucketKey = corruptBucketUsage(fsoBucketName, 777L, 88L, 3L);
+ applyQuotaRepair(request, 4L);
+
+ OmBucketInfo repaired = omMetadataManager.getBucketTable().get(bucketKey);
+ assertEquals(ratisBytes + ecBytes, repaired.getUsedBytes());
+ assertEquals(0L, repaired.getUsedNamespace());
+
+ abortMpu(fsoBucketName, keyName, uploadId,
BucketLayout.FILE_SYSTEM_OPTIMIZED, 5001L);
+ abortMpu(fsoBucketName, ecKeyName, ecUploadId,
BucketLayout.FILE_SYSTEM_OPTIMIZED, 5002L);
+ assertEquals(0L,
omMetadataManager.getBucketTable().get(bucketKey).getUsedBytes());
+ }
+
+ @Test
+ public void testQuotaRepairSkipsMPUOutsideRequestedBuckets() throws
Exception {
+ AtomicReference<OzoneManagerProtocolProtos.OMRequest> request =
mockQuotaRepairRequest();
+ OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName,
+ omMetadataManager, BucketLayout.OBJECT_STORE);
+ String skippedBucketName = "skipped" + bucketName;
+ OMRequestTestUtils.addBucketToDB(volumeName, skippedBucketName,
+ omMetadataManager, BucketLayout.OBJECT_STORE);
+
+ String repairedKey = "repairedMpuKey";
+ String repairedUploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo repairedInfo = newMultipartInfo(repairedUploadId,
4001L, RATIS_ONE, false);
+ repairedInfo.addPartKeyInfo(createPart(bucketName, repairedKey,
repairedUploadId, 1, 500L));
+ addMultipartInfo(bucketName, repairedKey, repairedInfo, 1L);
+
+ String skippedKey = "skippedMpuKey";
+ String skippedUploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo skippedInfo = newMultipartInfo(skippedUploadId, 4002L,
RATIS_ONE, false);
+ skippedInfo.addPartKeyInfo(createPart(skippedBucketName, skippedKey,
skippedUploadId, 1, 900L));
+ addMultipartInfo(skippedBucketName, skippedKey, skippedInfo, 2L);
+
+ String bucketKey = corruptBucketUsage(bucketName, 12345L, 99L, 3L);
+ String skippedBucketKey = corruptBucketUsage(skippedBucketName, 54321L,
77L, 4L);
+ applyQuotaRepair(request, 5L, bucketKey);
+
+ assertEquals(500L,
omMetadataManager.getBucketTable().get(bucketKey).getUsedBytes());
+ // the unrequested bucket keeps its stored usage, its parts are not folded
in anywhere
+ OmBucketInfo skipped =
omMetadataManager.getBucketTable().get(skippedBucketKey);
+ assertEquals(54321L, skipped.getUsedBytes());
+ assertEquals(77L, skipped.getUsedNamespace());
+ }
+
+ @Test
+ public void testMultipartCountLeavesUnrequestedBucketRowUnread() {
+ Map<String, QuotaRepairTask.CountPair> mpuCountMap = new HashMap<>();
+ mpuCountMap.put(OzoneConsts.OM_KEY_PREFIX + volumeName +
OzoneConsts.OM_KEY_PREFIX + bucketName
+ + OzoneConsts.OM_KEY_PREFIX, new QuotaRepairTask.CountPair());
+ OMMetadataManager metadataManager = mock(OMMetadataManager.class);
+
+ QuotaRepairTask.extractMultipartCount(
+ OmMultipartUpload.getDbKey(volumeName, "other" + bucketName, "mpuKey",
UUID.randomUUID().toString()),
+ mpuCountMap, metadataManager);
+
+ // the bucket is filtered on the key, so the row is never fetched or
decoded
+ verifyNoInteractions(metadataManager);
+ }
+
+ @Test
+ public void testMultipartCountReadsRequestedBucketRow() throws Exception {
+ QuotaRepairTask.CountPair usage = new QuotaRepairTask.CountPair();
+ Map<String, QuotaRepairTask.CountPair> mpuCountMap = new HashMap<>();
+ mpuCountMap.put(OzoneConsts.OM_KEY_PREFIX + volumeName +
OzoneConsts.OM_KEY_PREFIX + bucketName
+ + OzoneConsts.OM_KEY_PREFIX, usage);
+
+ String keyName = "mpuKey";
+ String uploadId = UUID.randomUUID().toString();
+ OmMultipartKeyInfo multipartInfo = newMultipartInfo(uploadId, 4003L,
+ RatisReplicationConfig.getInstance(THREE), false);
+ multipartInfo.addPartKeyInfo(createPart(bucketName, keyName, uploadId, 1,
100L));
+ multipartInfo.addPartKeyInfo(createPart(bucketName, keyName, uploadId, 2,
200L));
+ String dbKey = OmMultipartUpload.getDbKey(volumeName, bucketName, keyName,
uploadId);
+
+ Table<String, OmMultipartKeyInfo> multipartInfoTable = mock(Table.class);
+ when(multipartInfoTable.getSkipCache(dbKey)).thenReturn(multipartInfo);
+ OMMetadataManager metadataManager = mock(OMMetadataManager.class);
+
when(metadataManager.getMultipartInfoTable()).thenReturn(multipartInfoTable);
+
+ QuotaRepairTask.extractMultipartCount(dbKey, mpuCountMap, metadataManager);
+
+ assertEquals((100L + 200L) * 3, usage.getSpace());
+ verify(multipartInfoTable).getSkipCache(dbKey);
+ }
+
+ private AtomicReference<OzoneManagerProtocolProtos.OMRequest>
mockQuotaRepairRequest() throws Exception {
+ OzoneManagerProtocolProtos.OMResponse response =
mock(OzoneManagerProtocolProtos.OMResponse.class);
+ when(response.getSuccess()).thenReturn(true);
+ OzoneManagerRatisServer ratisServer = mock(OzoneManagerRatisServer.class);
+ AtomicReference<OzoneManagerProtocolProtos.OMRequest> request = new
AtomicReference<>();
+ doAnswer(invocation -> {
+ request.set(invocation.getArgument(0,
OzoneManagerProtocolProtos.OMRequest.class));
+ return response;
+ }).when(ratisServer).submitRequest(any(), any(), anyLong());
+ when(ozoneManager.getOmRatisServer()).thenReturn(ratisServer);
+ return request;
+ }
+
+ private OmMultipartKeyInfo newMultipartInfo(String uploadId, long objectId,
+ ReplicationConfig replicationConfig, boolean splitParts) {
+ OmMultipartKeyInfo.Builder builder =
OMRequestTestUtils.createOmMultipartKeyInfo(
+ uploadId, Time.now(), HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.ONE, objectId).toBuilder()
+ .setReplicationConfig(replicationConfig);
+ if (splitParts) {
+
builder.setSchemaVersion(OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION);
+ }
+ return builder.build();
+ }
+
+ private void addMultipartInfo(String bucket, String keyName,
OmMultipartKeyInfo multipartInfo,
+ long transactionIndex) throws IOException {
+ OmKeyInfo omKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName,
bucket,
+ keyName, multipartInfo.getReplicationConfig()).build();
+ OMRequestTestUtils.addMultipartInfoToTable(false, omKeyInfo,
multipartInfo, transactionIndex,
+ omMetadataManager);
+ }
+
+ // repair scans a RocksDB checkpoint, so parts must be written to the table,
not the cache
+ private void putSplitPart(String bucket, String keyName, String uploadId,
int partNumber,
+ long dataSize) throws IOException {
+
omMetadataManager.getMultipartPartsTable().put(OmMultipartPartKey.of(uploadId,
partNumber),
+ createSplitPart(bucket, keyName, uploadId, partNumber, dataSize));
+ }
+
+ private String corruptBucketUsage(String bucket, long usedBytes, long
usedNamespace, long transactionIndex)
+ throws IOException {
+ String bucketKey = omMetadataManager.getBucketKey(volumeName, bucket);
+ OmBucketInfo corrupted =
omMetadataManager.getBucketTable().get(bucketKey).toBuilder()
+ .setUsedBytes(usedBytes).setUsedNamespace(usedNamespace).build();
+ omMetadataManager.getBucketTable().put(bucketKey, corrupted);
+ omMetadataManager.getBucketTable().addCacheEntry(
+ new CacheKey<>(bucketKey), CacheValue.get(transactionIndex,
corrupted));
+ return bucketKey;
+ }
+
+ private void
applyQuotaRepair(AtomicReference<OzoneManagerProtocolProtos.OMRequest> request,
+ long transactionIndex, String... buckets) throws Exception {
+ QuotaRepairTask quotaRepairTask = new QuotaRepairTask(ozoneManager);
+ assertTrue(awaitRepair(quotaRepairTask.repair(Arrays.asList(buckets))));
+ OMClientResponse response = new OMQuotaRepairRequest(request.get())
+ .validateAndUpdateCache(ozoneManager, transactionIndex);
+ BatchOperation batchOperation =
omMetadataManager.getStore().initBatchOperation();
+ ((OMQuotaRepairResponse) response).addToDBBatch(omMetadataManager,
batchOperation);
+ omMetadataManager.getStore().commitBatchOperation(batchOperation);
+ }
+
+ private void abortMpu(String bucket, String keyName, String uploadId,
BucketLayout layout,
+ long transactionIndex) throws Exception {
+ OzoneManagerProtocolProtos.OMRequest abortRequest =
OMRequestTestUtils.createAbortMPURequest(
+ volumeName, bucket, keyName, uploadId);
+ OzoneManagerProtocolProtos.OMRequest preExecuted;
+ OMClientResponse response;
+ if (layout == BucketLayout.FILE_SYSTEM_OPTIMIZED) {
+ preExecuted = new S3MultipartUploadAbortRequestWithFSO(abortRequest,
layout).preExecute(ozoneManager);
+ response = new S3MultipartUploadAbortRequestWithFSO(preExecuted, layout)
+ .validateAndUpdateCache(ozoneManager, transactionIndex);
+ } else {
+ preExecuted = new S3MultipartUploadAbortRequest(abortRequest,
layout).preExecute(ozoneManager);
+ response = new S3MultipartUploadAbortRequest(preExecuted, layout)
+ .validateAndUpdateCache(ozoneManager, transactionIndex);
+ }
+ assertTrue(response.getOMResponse().getSuccess());
+ }
+
+ private PartKeyInfo createPart(String bucket, String keyName, String
uploadId, int partNumber,
+ long dataSize) {
+ return PartKeyInfo.newBuilder().setPartNumber(partNumber)
+ .setPartName(OmMultipartUpload.getDbKey(volumeName, bucket, keyName,
uploadId) + "/" + partNumber)
+
.setPartKeyInfo(KeyInfo.newBuilder().setVolumeName(volumeName).setBucketName(bucket)
+
.setKeyName(keyName).setDataSize(dataSize).setCreationTime(Time.now()).setModificationTime(Time.now())
+ .setType(HddsProtos.ReplicationType.RATIS).setFactor(ONE).build())
+ .build();
+ }
+
+ private OmMultipartPartInfo createSplitPart(String bucket, String keyName,
String uploadId,
+ int partNumber, long dataSize) {
+ OmKeyInfo partKeyInfo = OMRequestTestUtils.createOmKeyInfo(volumeName,
bucket, keyName,
+ RatisReplicationConfig.getInstance(ONE)).setDataSize(dataSize)
+ .setObjectID(2000L + partNumber).setUpdateID(2000L + partNumber)
+ .addMetadata(OzoneConsts.ETAG, "etag-" + partNumber).build();
+ return OmMultipartPartInfo.from(
+ OmMultipartUpload.getDbKey(volumeName, bucket, keyName, uploadId) +
"/" + partNumber,
+ partNumber, partKeyInfo);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
