This is an automated email from the ASF dual-hosted git repository.
priyeshkaratha 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 bc7055ab43d HDDS-16363. Fix split-schema MPU size tracking in Recon
event handlers (#11181)
bc7055ab43d is described below
commit bc7055ab43d459d001eb27030c00390809ad3b65
Author: Priyesh Karatha <[email protected]>
AuthorDate: Fri Sep 18 10:55:57 2026 +0530
HDDS-16363. Fix split-schema MPU size tracking in Recon event handlers
(#11181)
Co-authored-by: Priyesh Karatha <[email protected]>
---
.../recon/tasks/MultipartInfoInsightHandler.java | 118 +++++++++++++++++++--
.../ozone/recon/tasks/OmTableInsightTask.java | 32 +++++-
.../ozone/recon/tasks/TestOmTableInsightTask.java | 116 ++++++++++++++++++--
3 files changed, 249 insertions(+), 17 deletions(-)
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/MultipartInfoInsightHandler.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/MultipartInfoInsightHandler.java
index e4439c5e8d3..6e561e17501 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/MultipartInfoInsightHandler.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/MultipartInfoInsightHandler.java
@@ -17,8 +17,11 @@
package org.apache.hadoop.ozone.recon.tasks;
+import static
org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_INFO_TABLE;
+
import java.io.IOException;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.commons.lang3.tuple.Triple;
@@ -52,18 +55,40 @@
* {@code uploadId/partNumber}.</li>
* </ul>
*
- * <p>The event handlers below only access {@code multipartInfoTable} events,
- * whose values embed parts only for the legacy schema. Split-schema part
- * sizes therefore cannot be accounted incrementally from these events (the
- * handler has no DB access and the event carries no part data).
- * So they are reconciled during the periodic reprocess in {@link
#getTableSizeAndCount(String, OMMetadataManager)},
- * which reads the split {@code multipartPartsTable} directly.
+ * <p>Split-schema part sizes are accounted incrementally, one
+ * {@code multipartPartsTable} event at a time (see {@link
#handlePartPutEvent},
+ * {@link #handlePartDeleteEvent}, {@link #handlePartUpdateEvent}), each O(1).
+ * {@code OmMultipartPartInfo} does not carry a replication config, so the
parent
+ * MPU's config is cached here, keyed by uploadId, from the {@code
multipartInfoTable}
+ * PUT/UPDATE events handled below; OM always writes {@code multipartInfoTable}
+ * before the corresponding {@code multipartPartsTable} row in the same batch,
so the
+ * cache entry is available by the time a part event for that upload is
processed.
+ * The cache is bounded (LRU) rather than explicitly evicted per upload, since
a
+ * part-delete event can follow an info-table delete for the same upload
within the
+ * same batch and still needs the cached config.
*/
public class MultipartInfoInsightHandler implements OmTableHandler {
private static final Logger LOG =
LoggerFactory.getLogger(MultipartInfoInsightHandler.class);
+ private static final int MAX_CACHED_SPLIT_SCHEMA_UPLOADS = 10_000;
+
+ /**
+ * uploadId -> parent replication config, for split-schema MPUs, used to
compute
+ * the replicated size of {@code multipartPartsTable} events in O(1). Bounded
+ * (LRU, access-order) so long-lived Recon processes do not grow this
unbounded;
+ * a cache miss (e.g. under eviction pressure) simply skips that part's size
+ * update, which is reconciled by the next {@link #getTableSizeAndCount}
reprocess.
+ */
+ private final Map<String, ReplicationConfig>
splitSchemaUploadReplicationConfig =
+ new LinkedHashMap<String, ReplicationConfig>(16, 0.75f, true) {
+ @Override
+ protected boolean removeEldestEntry(Map.Entry<String,
ReplicationConfig> eldest) {
+ return size() > MAX_CACHED_SPLIT_SCHEMA_UPLOADS;
+ }
+ };
+
/**
* Consumes the (unreplicated, replicated) size of a single multipart part.
*/
@@ -85,6 +110,7 @@ public void handlePutEvent(OMDBUpdateEvent<String, Object>
event, String tableNa
objectCountMap.computeIfPresent(getTableCountKeyFromTable(tableName),
(k, count) -> count + 1L);
applyLegacyPartSizes(multipartKeyInfo, tableName, unReplicatedSizeMap,
replicatedSizeMap, true);
+ cacheSplitSchemaReplicationConfig(multipartKeyInfo);
} else {
LOG.warn("Put event does not have the Multipart Key Info for {}.",
event.getKey());
}
@@ -129,11 +155,56 @@ public void handleUpdateEvent(OMDBUpdateEvent<String,
Object> event, String tabl
OmMultipartKeyInfo newMultipartKeyInfo = (OmMultipartKeyInfo)
event.getValue();
applyLegacyPartSizes(oldMultipartKeyInfo, tableName,
unReplicatedSizeMap, replicatedSizeMap, false);
applyLegacyPartSizes(newMultipartKeyInfo, tableName,
unReplicatedSizeMap, replicatedSizeMap, true);
+ // A split-schema MPU's multipartInfoTable row is rewritten
(PUT->UPDATE) on
+ // every part commit, so refresh/populate the cache here too.
+ cacheSplitSchemaReplicationConfig(newMultipartKeyInfo);
} else {
LOG.warn("Update event does not have the Multipart Key Info for {}.",
event.getKey());
}
}
+ /**
+ * Invoked for a PUT (part committed) event on the {@code
multipartPartsTable}.
+ * Adds this part's size to the {@code multipartInfoTable} totals in O(1).
+ */
+ public void handlePartPutEvent(OMDBUpdateEvent<OmMultipartPartKey,
OmMultipartPartInfo> event,
+ Map<String, Long> unReplicatedSizeMap, Map<String, Long>
replicatedSizeMap) {
+ if (event.getValue() != null) {
+ applyPartSize(event.getKey(), event.getValue(), unReplicatedSizeMap,
replicatedSizeMap, true);
+ } else {
+ LOG.warn("Put event does not have the Multipart Part Info for {}.",
event.getKey());
+ }
+ }
+
+ /**
+ * Invoked for a DELETE (part removed on abort/complete) event on the
+ * {@code multipartPartsTable}. Subtracts this part's size from the
+ * {@code multipartInfoTable} totals in O(1).
+ */
+ public void handlePartDeleteEvent(OMDBUpdateEvent<OmMultipartPartKey,
OmMultipartPartInfo> event,
+ Map<String, Long> unReplicatedSizeMap, Map<String, Long>
replicatedSizeMap) {
+ if (event.getValue() != null) {
+ applyPartSize(event.getKey(), event.getValue(), unReplicatedSizeMap,
replicatedSizeMap, false);
+ } else {
+ LOG.warn("Delete event does not have the Multipart Part Info for {}.",
event.getKey());
+ }
+ }
+
+ /**
+ * Invoked for an UPDATE (part re-uploaded with the same part number) event
on
+ * the {@code multipartPartsTable}. Subtracts the old size and adds the new
+ * size from/to the {@code multipartInfoTable} totals in O(1).
+ */
+ public void handlePartUpdateEvent(OMDBUpdateEvent<OmMultipartPartKey,
OmMultipartPartInfo> event,
+ Map<String, Long> unReplicatedSizeMap, Map<String, Long>
replicatedSizeMap) {
+ if (event.getValue() == null || event.getOldValue() == null) {
+ LOG.warn("Update event does not have the old/new Multipart Part Info for
{}.", event.getKey());
+ return;
+ }
+ applyPartSize(event.getKey(), event.getOldValue(), unReplicatedSizeMap,
replicatedSizeMap, false);
+ applyPartSize(event.getKey(), event.getValue(), unReplicatedSizeMap,
replicatedSizeMap, true);
+ }
+
/**
* This method is called by the reprocess method. It calculates the record
* counts for the multipart info table. Additionally, it computes the sizes
@@ -215,7 +286,7 @@ public Triple<Long, Long, Long> getTableSizeAndCount(String
tableName,
/**
* Adds (or subtracts) the sizes of a legacy MPU's embedded parts to the size
* maps. Split-schema MPUs carry no embedded parts and are ignored here
(their
- * sizes are reconciled during reprocess; see class Javadoc).
+ * sizes are accounted per-part via {@link #handlePartPutEvent} and friends).
*
* @param add {@code true} to add the part sizes (PUT / new value in UPDATE),
* {@code false} to subtract them (DELETE / old value in UPDATE).
@@ -228,6 +299,39 @@ private void applyLegacyPartSizes(OmMultipartKeyInfo
multipartKeyInfo, String ta
});
}
+ /**
+ * Caches the replication config of a split-schema MPU, keyed by uploadId,
for use
+ * by {@link #applyPartSize}. No-op for legacy-schema MPUs.
+ */
+ private void cacheSplitSchemaReplicationConfig(OmMultipartKeyInfo
multipartKeyInfo) {
+ if (isLegacySchema(multipartKeyInfo)) {
+ return;
+ }
+ splitSchemaUploadReplicationConfig.put(multipartKeyInfo.getUploadID(),
multipartKeyInfo.getReplicationConfig());
+ }
+
+ /**
+ * Adds (or subtracts) a single split-schema part's size to/from the
+ * {@code multipartInfoTable} size maps, using the parent MPU's replication
+ * config cached by {@link #cacheSplitSchemaReplicationConfig}. If the
parent's
+ * config is not cached (e.g. evicted under memory pressure), the update is
+ * skipped with a warning; it is reconciled by the next reprocess.
+ */
+ private void applyPartSize(OmMultipartPartKey partKey, OmMultipartPartInfo
partInfo,
+ Map<String, Long> unReplicatedSizeMap, Map<String, Long>
replicatedSizeMap, boolean add) {
+ ReplicationConfig replicationConfig =
splitSchemaUploadReplicationConfig.get(partKey.getUploadId());
+ if (replicationConfig == null) {
+ LOG.warn("No cached replication config for upload {}; skipping part size
update.", partKey.getUploadId());
+ return;
+ }
+ long partDataSize = partInfo.getDataSize();
+ long partReplicatedSize = QuotaUtil.getReplicatedSize(partDataSize,
replicationConfig);
+ updateSize(unReplicatedSizeMap,
getUnReplicatedSizeKeyFromTable(MULTIPART_INFO_TABLE), partDataSize, add,
+ "unreplicated");
+ updateSize(replicatedSizeMap,
getReplicatedSizeKeyFromTable(MULTIPART_INFO_TABLE), partReplicatedSize, add,
+ "replicated");
+ }
+
/**
* Invokes {@code consumer} with the (unreplicated, replicated) size of each
* embedded part of a legacy MPU. Does nothing for split-schema MPUs.
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
index a4853dbf31d..2e6e0e51548 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/OmTableInsightTask.java
@@ -19,6 +19,7 @@
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.DELETED_TABLE;
import static
org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_INFO_TABLE;
+import static
org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_PARTS_TABLE;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_FILE_TABLE;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_KEY_TABLE;
@@ -42,6 +43,8 @@
import org.apache.hadoop.hdds.utils.db.TableIterator;
import org.apache.hadoop.hdds.utils.db.cache.TableCache;
import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartInfo;
+import org.apache.hadoop.ozone.om.helpers.OmMultipartPartKey;
import org.apache.hadoop.ozone.recon.ReconServerConfigKeys;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
import org.apache.hadoop.ozone.recon.spi.ReconGlobalStatsManager;
@@ -61,6 +64,7 @@ public class OmTableInsightTask implements ReconOmTask {
private ReconGlobalStatsManager reconGlobalStatsManager;
private ReconOMMetadataManager reconOMMetadataManager;
private Map<String, OmTableHandler> tableHandlers;
+ private MultipartInfoInsightHandler multipartInfoInsightHandler;
private Collection<String> tables;
private Map<String, Long> objectCountMap;
private Map<String, Long> unReplicatedSizeMap;
@@ -79,7 +83,8 @@ public OmTableInsightTask(ReconGlobalStatsManager
reconGlobalStatsManager,
tableHandlers.put(OPEN_KEY_TABLE, new OpenKeysInsightHandler());
tableHandlers.put(OPEN_FILE_TABLE, new OpenKeysInsightHandler());
tableHandlers.put(DELETED_TABLE, new DeletedKeysInsightHandler());
- tableHandlers.put(MULTIPART_INFO_TABLE, new MultipartInfoInsightHandler());
+ multipartInfoInsightHandler = new MultipartInfoInsightHandler();
+ tableHandlers.put(MULTIPART_INFO_TABLE, multipartInfoInsightHandler);
this.maxKeysInMemory =
reconOMMetadataManager.getOzoneConfiguration().getInt(
ReconServerConfigKeys.OZONE_RECON_TASK_REPROCESS_MAX_KEYS_IN_MEMORY,
ReconServerConfigKeys.OZONE_RECON_TASK_REPROCESS_MAX_KEYS_IN_MEMORY_DEFAULT);
@@ -315,6 +320,10 @@ private void handlePutEvent(OMDBUpdateEvent<String,
Object> event,
} else {
String countKey = getTableCountKeyFromTable(tableName);
objectCountMap.computeIfPresent(countKey, (k, count) -> count + 1L);
+ if (tableName.equals(MULTIPART_PARTS_TABLE)) {
+
multipartInfoInsightHandler.handlePartPutEvent(castToPartEvent(event),
+ unReplicatedSizeMap, replicatedSizeMap);
+ }
}
}
}
@@ -329,6 +338,10 @@ private void handleDeleteEvent(OMDBUpdateEvent<String,
Object> event,
} else {
objectCountMap.computeIfPresent(getTableCountKeyFromTable(tableName),
(k, count) -> count > 0 ? count - 1L : 0L);
+ if (tableName.equals(MULTIPART_PARTS_TABLE)) {
+
multipartInfoInsightHandler.handlePartDeleteEvent(castToPartEvent(event),
+ unReplicatedSizeMap, replicatedSizeMap);
+ }
}
}
}
@@ -342,10 +355,27 @@ private void handleUpdateEvent(OMDBUpdateEvent<String,
Object> event,
// Handle update for only size related tables
tableHandler.handleUpdateEvent(event, tableName, objectCountMap,
unReplicatedSizeMap, replicatedSizeMap);
+ } else if (tableName.equals(MULTIPART_PARTS_TABLE)) {
+
multipartInfoInsightHandler.handlePartUpdateEvent(castToPartEvent(event),
+ unReplicatedSizeMap, replicatedSizeMap);
}
}
}
+ /**
+ * The multipartPartsTable is not registered in {@link #tableHandlers}, so
its events fall
+ * through to the generic {@code tableHandler == null} branch above for
object counting, and are
+ * additionally routed here to update multipartInfoTable's sizes. Its
keys/values are genuinely
+ * {@link OmMultipartPartKey}/{@link OmMultipartPartInfo} at runtime (see
+ * {@code OMDBUpdatesHandler}); erasure makes this cast safe despite the
static
+ * {@code OMDBUpdateEvent<String, Object>} type of {@code event}.
+ */
+ @SuppressWarnings("unchecked")
+ private static OMDBUpdateEvent<OmMultipartPartKey, OmMultipartPartInfo>
castToPartEvent(
+ OMDBUpdateEvent<String, Object> event) {
+ return (OMDBUpdateEvent<OmMultipartPartKey, OmMultipartPartInfo>)
(OMDBUpdateEvent<?, ?>) event;
+ }
+
/**
* Write the updated count and size information to the database.
*
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
index ef44075d80a..9b5f3b0c1f9 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/tasks/TestOmTableInsightTask.java
@@ -22,6 +22,7 @@
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.DELETED_TABLE;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.KEY_TABLE;
import static
org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_INFO_TABLE;
+import static
org.apache.hadoop.ozone.om.codec.OMDBDefinition.MULTIPART_PARTS_TABLE;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_FILE_TABLE;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.OPEN_KEY_TABLE;
import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.VOLUME_TABLE;
@@ -492,8 +493,9 @@ public void testProcessForCount() {
// Creating events for each table except the deleted table
for (String tableName : omTableInsightTask.getTaskTables()) {
- if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)) {
- continue; // Skipping deleted and multipartInfo tables as they have
separate tests
+ if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)
+ || tableName.equals(MULTIPART_PARTS_TABLE)) {
+ continue; // Skipping deleted, multipartInfo and multipartParts tables
as they have separate tests
}
// Adding 5 PUT events per table
@@ -520,7 +522,8 @@ public void testProcessForCount() {
// Verifying the count in each table
for (String tableName : omTableInsightTask.getTaskTables()) {
- if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)) {
+ if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)
+ || tableName.equals(MULTIPART_PARTS_TABLE)) {
continue;
}
assertEquals(4L, getCountForTable(
@@ -530,7 +533,8 @@ public void testProcessForCount() {
List<OMDBUpdateEvent> additionalEvents = new ArrayList<>();
// Simulating new PUT and DELETE events
for (String tableName : omTableInsightTask.getTaskTables()) {
- if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)) {
+ if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)
+ || tableName.equals(MULTIPART_PARTS_TABLE)) {
continue;
}
// Adding 1 new PUT event
@@ -548,7 +552,8 @@ public void testProcessForCount() {
omTableInsightTask.process(additionalBatch, Collections.emptyMap());
// Verifying the final count in each table
for (String tableName : omTableInsightTask.getTaskTables()) {
- if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)) {
+ if (tableName.equals(DELETED_TABLE) ||
tableName.equals(MULTIPART_INFO_TABLE)
+ || tableName.equals(MULTIPART_PARTS_TABLE)) {
continue;
}
// 5 items expected after processing the additional events.
@@ -750,6 +755,89 @@ public void testProcessForMultipartInfoTable() {
assertEquals(2700L, getReplicatedSizeForTable(MULTIPART_INFO_TABLE));
}
+ /**
+ * Validates the event-driven (O(1) per part event) accounting of
split-schema MPU sizes:
+ * multipartInfoTable PUT caches the parent's replication config,
multipartPartsTable PUT/UPDATE/DELETE
+ * events adjust the multipartInfoTable size totals incrementally, and a
part re-upload (UPDATE, same
+ * part number) nets the old size out and the new size in rather than
cancelling to zero.
+ */
+ @Test
+ public void testProcessForSplitSchemaMPU() throws Exception {
+ String uploadID = UUID.randomUUID().toString();
+ String volumeName = UUID.randomUUID().toString();
+ String bucketName = UUID.randomUUID().toString();
+ String keyName = UUID.randomUUID().toString();
+
+ OmMultipartKeyInfo splitMpu = new OmMultipartKeyInfo.Builder()
+ .setObjectID(1L)
+ .setUploadID(uploadID)
+ .setCreationTime(Time.now())
+ .setReplicationConfig(RatisReplicationConfig.getInstance(
+ HddsProtos.ReplicationFactor.THREE))
+ .setSchemaVersion(OmMultipartKeyInfo.SPLIT_PARTS_TABLE_SCHEMA_VERSION)
+ .build();
+ String multipartKey = reconOMMetadataManager.getMultipartKey(volumeName,
bucketName, keyName, uploadID);
+
+ // MPU initiation: caches the parent's replication config for the part
events below.
+ ArrayList<OMDBUpdateEvent> initEvents = new ArrayList<>();
+ initEvents.add(getOMUpdateEvent(multipartKey, splitMpu,
MULTIPART_INFO_TABLE, PUT, null));
+ omTableInsightTask.process(new OMUpdateEventBatch(initEvents, 0L),
Collections.emptyMap());
+
+ assertEquals(1L, getCountForTable(MULTIPART_INFO_TABLE));
+ assertEquals(0L, getUnReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+ assertEquals(0L, getReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+
+ // 3 parts uploaded (multipartPartsTable PUT), 100 bytes each.
+ OmMultipartPartInfo part1 = buildSplitSchemaPartInfo(volumeName,
bucketName, keyName, uploadID, 1, 100L);
+ OmMultipartPartInfo part2 = buildSplitSchemaPartInfo(volumeName,
bucketName, keyName, uploadID, 2, 100L);
+ OmMultipartPartInfo part3 = buildSplitSchemaPartInfo(volumeName,
bucketName, keyName, uploadID, 3, 100L);
+ ArrayList<OMDBUpdateEvent> partPutEvents = new ArrayList<>();
+ partPutEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 1),
part1, MULTIPART_PARTS_TABLE, PUT, null));
+ partPutEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 2),
part2, MULTIPART_PARTS_TABLE, PUT, null));
+ partPutEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 3),
part3, MULTIPART_PARTS_TABLE, PUT, null));
+ omTableInsightTask.process(new OMUpdateEventBatch(partPutEvents, 0L),
Collections.emptyMap());
+
+ // 3 parts * 100 bytes = 300 bytes unreplicated; RATIS THREE -> 900 bytes
replicated.
+ assertEquals(1L, getCountForTable(MULTIPART_INFO_TABLE));
+ assertEquals(300L, getUnReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+ assertEquals(900L, getReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+
+ // Part 1 re-uploaded (multipartPartsTable UPDATE) with the same part
number, 100 -> 200 bytes.
+ OmMultipartPartInfo newPart1 = buildSplitSchemaPartInfo(volumeName,
bucketName, keyName, uploadID, 1, 200L);
+ ArrayList<OMDBUpdateEvent> partUpdateEvents = new ArrayList<>();
+ partUpdateEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 1),
newPart1, MULTIPART_PARTS_TABLE,
+ UPDATE, part1));
+ omTableInsightTask.process(new OMUpdateEventBatch(partUpdateEvents, 0L),
Collections.emptyMap());
+
+ // 200 + 100 + 100 = 400 bytes unreplicated; RATIS THREE -> 1200 bytes
replicated.
+ assertEquals(400L, getUnReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+ assertEquals(1200L, getReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+
+ // Part 3 removed (multipartPartsTable DELETE).
+ ArrayList<OMDBUpdateEvent> partDeleteEvents = new ArrayList<>();
+ partDeleteEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 3),
part3, MULTIPART_PARTS_TABLE,
+ DELETE, null));
+ omTableInsightTask.process(new OMUpdateEventBatch(partDeleteEvents, 0L),
Collections.emptyMap());
+
+ // 200 + 100 = 300 bytes unreplicated; RATIS THREE -> 900 bytes replicated.
+ assertEquals(300L, getUnReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+ assertEquals(900L, getReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+
+ // MPU aborted: remaining parts removed, followed by the
multipartInfoTable DELETE, matching OM's
+ // write-ordering guarantee that multipartInfoTable writes/deletes precede
multipartPartsTable ones.
+ ArrayList<OMDBUpdateEvent> finalEvents = new ArrayList<>();
+ finalEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 1),
newPart1, MULTIPART_PARTS_TABLE,
+ DELETE, null));
+ finalEvents.add(getOMUpdateEvent(OmMultipartPartKey.of(uploadID, 2),
part2, MULTIPART_PARTS_TABLE,
+ DELETE, null));
+ finalEvents.add(getOMUpdateEvent(multipartKey, splitMpu,
MULTIPART_INFO_TABLE, DELETE, null));
+ omTableInsightTask.process(new OMUpdateEventBatch(finalEvents, 0L),
Collections.emptyMap());
+
+ assertEquals(0L, getCountForTable(MULTIPART_INFO_TABLE));
+ assertEquals(0L, getUnReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+ assertEquals(0L, getReplicatedSizeForTable(MULTIPART_INFO_TABLE));
+ }
+
@Test
public void testReprocessForMultipartInfoTable() throws Exception {
String uploadID = UUID.randomUUID().toString();
@@ -890,6 +978,18 @@ public void
testReprocessForMultipartInfoTableMixedSchema() throws Exception {
*/
private void putSplitSchemaPart(String uploadID, String volumeName, String
bucketName,
String keyName, int partNumber, long dataSize) throws IOException {
+ OmMultipartPartInfo partInfo =
+ buildSplitSchemaPartInfo(volumeName, bucketName, keyName, uploadID,
partNumber, dataSize);
+ reconOMMetadataManager.getMultipartPartsTable().put(
+ OmMultipartPartKey.of(uploadID, partNumber), partInfo);
+ }
+
+ /**
+ * Builds a single split-schema part (an {@link OmMultipartPartInfo}),
without writing it to the
+ * multipartPartsTable. An ETag is set because it is mandatory for
split-schema parts.
+ */
+ private OmMultipartPartInfo buildSplitSchemaPartInfo(String volumeName,
String bucketName,
+ String keyName, String uploadID, int partNumber, long dataSize) {
OmKeyInfo partOmKeyInfo = new OmKeyInfo.Builder()
.setVolumeName(volumeName)
.setBucketName(bucketName)
@@ -907,9 +1007,7 @@ private void putSplitSchemaPart(String uploadID, String
volumeName, String bucke
.addMetadata(OzoneConsts.ETAG, "etag-" + partNumber)
.build();
String partName = reconOMMetadataManager.getMultipartKey(volumeName,
bucketName, keyName, uploadID);
- OmMultipartPartInfo partInfo = OmMultipartPartInfo.from(partName,
partNumber, partOmKeyInfo);
- reconOMMetadataManager.getMultipartPartsTable().put(
- OmMultipartPartKey.of(uploadID, partNumber), partInfo);
+ return OmMultipartPartInfo.from(partName, partNumber, partOmKeyInfo);
}
public PartKeyInfo createPartKeyInfo(String volumeName, String bucketName,
@@ -933,7 +1031,7 @@ public PartKeyInfo createPartKeyInfo(String volumeName,
String bucketName,
}
private OMDBUpdateEvent getOMUpdateEvent(
- String name, Object value,
+ Object name, Object value,
String table,
OMDBUpdateEvent.OMDBUpdateAction action,
Object oldValue) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
