voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3956128202
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java:
##########
@@ -189,29 +193,54 @@ public String overwriteHoodieProperties(
@ShellMethod(key = "repair corrupted clean files", value = "repair corrupted
clean files")
public void removeCorruptedPendingCleanAction() {
+ removeCorruptedPendingCleanAction(HoodieCLI.getTableMetaClient());
+ }
- HoodieTableMetaClient client = HoodieCLI.getTableMetaClient();
- HoodieTimeline cleanerTimeline =
HoodieCLI.getTableMetaClient().getActiveTimeline().getCleanerTimeline();
+ /**
+ * Removes the pending clean instants whose plan is verifiably empty or
corrupt.
+ * <p>
+ * The plan bytes are read in full before anything is judged. The timeline
serde wraps every
+ * exception raised while it streams an instant file, a transient read
failure included, in
+ * the same "unable to read commit metadata" IOException that an empty or
truncated file
+ * raises, so the message cannot tell a storage outage from corruption. A
failure of the
+ * read itself is therefore propagated, and only the in-memory decode, which
no I/O can
+ * disturb, decides that the plan is corrupt.
+ */
+ static void removeCorruptedPendingCleanAction(HoodieTableMetaClient client) {
+ HoodieActiveTimeline activeTimeline = client.getActiveTimeline();
+ HoodieTimeline cleanerTimeline = activeTimeline.getCleanerTimeline();
log.info("Inspecting pending clean metadata in timeline for corrupted
files");
cleanerTimeline.filterInflightsAndRequested().getInstants().forEach(instant -> {
- try {
- CleanerUtils.getCleanerPlan(client, instant);
- } catch (AvroRuntimeException e) {
- log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
- TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
- instant, client.getInstantFileNameGenerator());
- } catch (IOException ioe) {
- if (ioe.getMessage().contains("Not an Avro data file")) {
- log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
- TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
- instant, client.getInstantFileNameGenerator());
- } else {
- throw new HoodieIOException(ioe.getMessage(), ioe);
- }
+ HoodieInstant planInstant = CleanerUtils.getCleanRequestInstant(client,
instant);
+ byte[] plan;
+ try (InputStream in =
activeTimeline.getInstantContentStream(planInstant)) {
+ plan = FileIOUtils.readAsByteArray(in);
+ } catch (IOException e) {
+ throw new HoodieIOException("Failed to read the plan of pending clean
instant " + instant, e);
+ }
+ if (plan.length > 0 && isReadableCleanerPlan(client, plan)) {
+ return;
}
+ log.warn("Corruption found. Trying to remove corrupted clean instant
file: {}", instant);
+ TimelineUtils.deleteInstantFile(client.getStorage(),
client.getTimelinePath(),
+ instant, client.getInstantFileNameGenerator());
});
}
+ /**
+ * Decodes a clean plan held in memory, which fails only on the content
itself: bytes that are not
+ * an Avro file or that stop short fail the read, and an Avro container that
holds no record, which
+ * a writer killed between opening and closing it leaves behind, fails the
serde's argument check.
+ */
+ private static boolean isReadableCleanerPlan(HoodieTableMetaClient client,
byte[] plan) {
+ try {
+ CleanerUtils.getCleanerPlan(client, new ByteArrayInputStream(plan));
+ return true;
+ } catch (IOException | AvroRuntimeException | IllegalArgumentException e) {
Review Comment:
Confirmed, and the default is worse than the ceiling case.
`SystemLimitException` itself does extend `AvroRuntimeException`, but its three
check methods throw a bare `UnsupportedOperationException` above
`MAX_ARRAY_VM_LIMIT`, and `maxBytesLength` / `maxCollectionLength` /
`maxStringLength` all initialise to `MAX_ARRAY_VM_LIMIT` too -- so unless
someone sets the system properties, the `SystemLimitException` branch is
unreachable and an oversized length out of `BinaryDecoder.readString` /
`readBytes` / `doReadItemCount` always lands on the
`UnsupportedOperationException`.
The version unboxing holds as well: `HoodieCleanerPlan.avsc` declares
`version` as `["int","null"]`, so `getVersion()` is an `Integer` handed to
`MetadataMigrator.upgradeToLatest(T, int)`.
Widened to `IOException | RuntimeException`. Your safety argument checks out
-- the bytes are already in memory and both `CleanPlanMigrationHandler`s are
pure transforms, so there is no I/O failure in there to swallow.
`testRemoveCorruptedPendingCleanActionRepairsPastAnUndecodablePlan` writes a
null-version plan ahead of an ordinary corruption and asserts the instant
behind it is still repaired.
--
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]
