rakeshadr commented on code in PR #10849:
URL: https://github.com/apache/ozone/pull/10849#discussion_r3637546895
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java:
##########
@@ -1138,8 +1138,15 @@ private void processMultipartUploads(OmBucketInfo
bucketInfo, List<OmLCRule> rul
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());
+ // Get part count for this MPU (at least 1 even if no parts
uploaded yet).
+ // Split-schema MPUs keep parts in the separate
multipartPartsTable
+ // (the embedded map is empty), so count those rows; legacy MPUs
use
+ // the embedded map.
+ int uploadedParts = mpuKeyInfo.getSchemaVersion()
+ == OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION
+ ? OMMultipartUploadUtils.countParts(omMetadataManager,
upload.getUploadId())
Review Comment:
Error for ONE bad upload's part prefix propagates IOException out of the
for-loop and break. Can you wrap the `#countParts()` call in its own
`try/catch`. This would allow loop to continue for next MPU.
```
try {
uploadedParts = mpuKeyInfo.getSchemaVersion()
== OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION
? OMMultipartUploadUtils.countParts(omMetadataManager,
upload.getUploadId())
: mpuKeyInfo.getPartKeyInfoMap().size();
} catch (IOException e) {
LOG.warn("Failed to count parts for MPU {}/{}/{} uploadId {}, skipping",
volumeName, bucketName, keyName, upload.getUploadId(), e);
break; // break out of the ruleList loop, continue outer while
}
```
##########
hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/om/ContainerToKeyMapping.java:
##########
@@ -313,11 +318,18 @@ private void processMultipartUpload(Set<Long>
containerIds, Map<Long, List<Strin
String dbKey = entry.getKey();
OmMultipartKeyInfo mpuInfo = entry.getValue();
- // Collect all target containers that have parts of this MPU
+ // Collect all target containers that have parts of this MPU. Legacy
+ // (schemaVersion 0) MPUs embed their parts in the multipartInfoTable
+ // value, whereas split (schemaVersion 1) MPUs store each part as a
+ // separate row in the multipartPartsTable.
Set<Long> matchedContainers = new HashSet<>();
- for (PartKeyInfo partKeyInfo : mpuInfo.getPartKeyInfoMap()) {
- OmKeyInfo partKey =
OmKeyInfo.getFromProtobuf(partKeyInfo.getPartKeyInfo());
- matchedContainers.addAll(getKeyContainers(partKey, containerIds));
+ if (mpuInfo.getSchemaVersion() ==
OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION) {
+ matchedContainers.addAll(getSplitPartContainers(dbKey,
containerIds));
Review Comment:
Please add test for the new code path. Thanks!
TestOmMetadataManager
TestContainerToKeyMapping
TestKeyLifecycleService
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyLifecycleService.java:
##########
@@ -1138,8 +1138,15 @@ private void processMultipartUploads(OmBucketInfo
bucketInfo, List<OmLCRule> rul
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());
+ // Get part count for this MPU (at least 1 even if no parts
uploaded yet).
+ // Split-schema MPUs keep parts in the separate
multipartPartsTable
+ // (the embedded map is empty), so count those rows; legacy MPUs
use
+ // the embedded map.
+ int uploadedParts = mpuKeyInfo.getSchemaVersion()
Review Comment:
can you add test for MPU part count failure case as well.
Note: LLM generated unit test, please use it as reference and validate it
before adding to the PR.
```
@Test
public void testPartCountIoExceptionSkipsUploadContinuesBucket() throws
Exception {
// Upload A: valid, 2 parts in multipartPartsTable
// Upload B: corrupt — no table entry but we simulate IOException via mock
// Upload C: valid, 1 part
// Expectation: A and C are added to expiredUploads; B is skipped with warn
OMMetadataManager mockMM = Mockito.spy(omMetadataManager);
String uploadA = OMMultipartUploadUtils.getMultipartUploadId();
String uploadB = OMMultipartUploadUtils.getMultipartUploadId();
// Write 2 parts for A
for (int p = 1; p <= 2; p++) {
omMetadataManager.getMultipartPartsTable().put(
OmMultipartPartKey.of(uploadA, p),
new OmMultipartPartInfo.Builder()
.setPartName("a/p" + p).setPartNumber(p)
.setDataSize(100L).setObjectID(p).setUpdateID(p).build());
}
// Simulate IOException when accessing parts for uploadB
Table<OmMultipartPartKey, OmMultipartPartInfo> mockPartsTable =
Mockito.spy(omMetadataManager.getMultipartPartsTable());
Mockito.when(mockMM.getMultipartPartsTable()).thenReturn(mockPartsTable);
Mockito.when(mockPartsTable.iterator(
Mockito.eq(OmMultipartPartKey.prefix(uploadB))))
.thenThrow(new IOException("simulated corruption"));
assertEquals(2, OMMultipartUploadUtils.countParts(mockMM, uploadA));
// countParts for B must throw, not propagate bucket-level
assertThrows(IOException.class,
() -> OMMultipartUploadUtils.countParts(mockMM, uploadB));
// The fix: bucket processing does NOT stop when B throws
KeyLifecycleService.PartCountLimitedList list =
new KeyLifecycleService.PartCountLimitedList(10);
for (String uid : List.of(uploadA, uploadB)) {
try {
int c = OMMultipartUploadUtils.countParts(mockMM, uid);
list.add(new OmMultipartUpload("v", "b", "k", uid), Math.max(1, c));
} catch (IOException e) {
// per-MPU skip — bucket loop continues
}
}
assertEquals(1, list.size()); // only A; B was skipped
assertEquals(2, list.getPartCount());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
