wombatu-kun commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3953752238
##########
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:
Avro routes every string, bytes and collection length through
SystemLimitException, which throws UnsupportedOperationException - not
AvroRuntimeException - once a corrupt length passes the VM array limit, and
CleanerUtils.getCleanerPlan also unboxes the schema-nullable version field, so
either one escapes this catch and aborts the repair for every later instant.
Widening it to RuntimeException is safe here: both CleanPlanMigrationHandlers
are pure in-memory transforms, so there is no I/O failure to swallow.
##########
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(),
Review Comment:
The plan is read from getCleanRequestInstant but the delete is handed the
original instant, and TimelineUtils.deleteInstantFile removes only that
instant's own file, so an inflight clean with a corrupt requested plan loses
its inflight file and keeps the corrupt one. Is the second pass intentional, or
should the delete take the requested instant too?
--
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]
