HDDS-15949. Handle split MPU part counting for abort batch sizing and container-key-mapping CLI

[spacemonkd]

What changes were proposed in this pull request?

HDDS-15949. Handle split MPU part counting for abort batch sizing and container-key-mapping CLI

Please describe your PR in detail:

Schema-aware MPU part counting for abort batch sizing (OmMetadataManagerImpl.getExpiredMultipartUploads(), KeyLifecycleService)

For split-schema MPUs (schemaVersion == 1), part count now comes from multipartPartsTable via OMMultipartUploadUtils.countParts() instead of getPartKeyInfoMap().size() (which is always 0 for split MPUs).

Without fix the expired MPU cleanup and lifecycle AbortIncompleteMultipartUpload treated every split MPU as having 0–1 parts when sizing abort batches. Batches could exceed the intended part limit (maxParts / mpuAbortLimitPerTask), causing larger-than-planned abort work per iteration. The abort logic itself was already schema-aware; only batch sizing was wrong.

Shared countParts() helper (OMMultipartUploadUtils)

Added countParts(omMetadataManager, uploadId) wrapping the existing cache-aware scanParts() logic.
Without fix it is the same as above, plus duplicated/incorrect counting if each call site reimplemented split-table reads differently from abort/complete paths.

Schema-aware split MPU handling in debug CLI (ContainerToKeyMapping)

processMultipartUpload branches on schema version:

• legacy MPUs still use embedded PartKeyInfo
• split MPUs scan multipartPartsTable by uploadId and map block containers from each part’s locations.
  Without fix the ozone debug om container-key-mapping --in-progress reported no open MPU keys for split-schema uploads, even when their parts used the queried containers, this causes misleading output for debugging container to key mappings on upgraded clusters.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-15949

How was this patch tested?

Patch was tested via unit tests

[rakeshadr]

Thanks @spacemonkd for handling this case. Added a few comments, pls check it.

[rakeshadr]

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
}

[rakeshadr]

Please add test for the new code path. Thanks!

TestOmMetadataManager
TestContainerToKeyMapping
TestKeyLifecycleService

[rakeshadr]

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());
}

[sarvekshayr]

Thanks @spacemonkd for working on this.

[sarvekshayr]

Use the below approach instead of manual substring parsing -

Suggested change

	String uploadId = multipartInfoDbKey.substring(
	multipartInfoDbKey.lastIndexOf(OM_KEY_PREFIX) + OM_KEY_PREFIX.length());
	String uploadId;
	try {
	uploadId = OmMultipartUpload.from(multipartInfoDbKey).getUploadId();
	} catch (IllegalArgumentException e) {
	err().println("Invalid multipartInfoTable key " + multipartInfoDbKey + ", " + e);
	return matchedContainers;
	}

[rakeshadr]

Added a few minor comments. Please look at other comments and resolve once its fixed. Thanks!

[rakeshadr]

Not a bug, but using a distinct CONTAINER_ID_5 for the new tests would make the split-schema test self-contained and easier to reason about in future.

CONTAINER_ID_4 for both legacy and split MPU testContainerToKeyMappingWithSplitSchemaMPU

[rakeshadr]

Can you please add comments to help maintainers.

 // A (2 parts) and C (1 part) are added; B is skipped due to IOException
 assertEquals(2, list.size()); // uploadA and uploadC only
 assertEquals(3, list.getPartCount()); // 2 (uploadA) + 1 (uploadC)