voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3949859339
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java:
##########
@@ -156,94 +345,77 @@ public String showArchivedCommits(
allStats.addAll(readCommits);
}
}
- TableHeader header = new
TableHeader().addTableHeaderField("action").addTableHeaderField("instant")
-
.addTableHeaderField("partition").addTableHeaderField("file_id").addTableHeaderField("prev_instant")
-
.addTableHeaderField("num_writes").addTableHeaderField("num_inserts").addTableHeaderField("num_deletes")
-
.addTableHeaderField("num_update_writes").addTableHeaderField("total_log_files")
-
.addTableHeaderField("total_log_blocks").addTableHeaderField("total_corrupt_log_blocks")
-
.addTableHeaderField("total_rollback_blocks").addTableHeaderField("total_log_records")
-
.addTableHeaderField("total_updated_records_compacted").addTableHeaderField("total_write_bytes")
- .addTableHeaderField("total_write_errors");
-
- return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allStats);
+ return allStats;
}
- @ShellMethod(key = "show archived commits", value = "Read commits from
archived files and show details")
- public String showCommits(
- @ShellOption(value = {"--skipMetadata"}, help = "Skip displaying commit
metadata",
- defaultValue = "true") boolean skipMetadata,
- @ShellOption(value = {"--limit"}, help = "Limit commits", defaultValue =
"10") final Integer limit,
- @ShellOption(value = {"--sortBy"}, help = "Sorting Field", defaultValue
= "") final String sortByField,
- @ShellOption(value = {"--desc"}, help = "Ordering", defaultValue =
"false") final boolean descending,
- @ShellOption(value = {"--headeronly"}, help = "Print Header Only",
- defaultValue = "false") final boolean headerOnly)
- throws IOException {
-
- System.out.println("===============> Showing only " + limit + " archived
commits <===============");
- HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
- StoragePath archivePath =
- new StoragePath(metaClient.getArchivePath(), ".commits_.archive*");
- HoodieStorage storage = metaClient.getStorage();
- List<StoragePathInfo> pathInfoList = storage.globEntries(archivePath);
- List<Comparable[]> allCommits = new ArrayList<>();
- for (StoragePathInfo pathInfo : pathInfoList) {
- // read the archived file
- try (HoodieLogFormat.Reader reader =
HoodieLogFormat.newReader(metaClient,
- new HoodieLogFile(pathInfo.getPath()),
HoodieSchema.fromAvroSchema(HoodieArchivedMetaEntry.getClassSchema()))) {
- List<IndexedRecord> readRecords = new ArrayList<>();
- // read the avro blocks
- while (reader.hasNext()) {
- HoodieAvroDataBlock blk = (HoodieAvroDataBlock) reader.next();
- try (ClosableIterator<HoodieRecord<IndexedRecord>> recordItr =
blk.getRecordIterator(HoodieRecordType.AVRO)) {
- recordItr.forEachRemaining(r -> readRecords.add(r.getData()));
- }
- }
- List<Comparable[]> readCommits = readRecords.stream().map(r ->
(GenericRecord) r)
- .map(r -> readCommit(r,
skipMetadata)).collect(Collectors.toList());
- allCommits.addAll(readCommits);
- }
- }
-
- TableHeader header = new
TableHeader().addTableHeaderField("CommitTime").addTableHeaderField("CommitType");
-
- if (!skipMetadata) {
- header = header.addTableHeaderField("CommitDetails");
- }
-
- return HoodiePrintHelper.print(header, new HashMap<>(), sortByField,
descending, limit, headerOnly, allCommits);
+ private static boolean isCompletedEntry(GenericRecord archivedEntry) {
+ // entries written before the state was recorded are completed ones
+ Object actionState = archivedEntry.get("actionState");
+ return actionState == null ||
HoodieInstant.State.COMPLETED.name().equals(actionState.toString());
}
- private Comparable[] commitDetail(GenericRecord record, String metadataName,
boolean skipMetadata) {
- List<Object> commitDetails = new ArrayList<>();
- commitDetails.add(record.get("commitTime"));
- commitDetails.add(record.get("actionType").toString());
+ private static Comparable[] readArchivedCommit(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant,
+ boolean skipMetadata, boolean
legacyArchive) {
+ List<Comparable> commitDetails = new ArrayList<>();
+ commitDetails.add(instant.requestedTime());
+ commitDetails.add(instant.getAction());
if (!skipMetadata) {
-
commitDetails.add(Option.ofNullable(record.get(metadataName)).orElse("{}").toString());
+ commitDetails.add(readArchivedMetadataString(archivedTimeline, instant,
legacyArchive));
}
return commitDetails.toArray(new Comparable[commitDetails.size()]);
}
- private Comparable[] readCommit(GenericRecord record, boolean skipMetadata) {
- String actionType = record.get("actionType").toString();
- switch (actionType) {
- case HoodieTimeline.CLEAN_ACTION:
- return commitDetail(record, "hoodieCleanMetadata", skipMetadata);
- case HoodieTimeline.COMMIT_ACTION:
- case HoodieTimeline.DELTA_COMMIT_ACTION:
- return commitDetail(record, "hoodieCommitMetadata", skipMetadata);
- case HoodieTimeline.ROLLBACK_ACTION:
- return commitDetail(record, "hoodieRollbackMetadata", skipMetadata);
- case HoodieTimeline.SAVEPOINT_ACTION:
- return commitDetail(record, "hoodieSavePointMetadata", skipMetadata);
- case HoodieTimeline.COMPACTION_ACTION:
- return commitDetail(record, "hoodieCompactionMetadata", skipMetadata);
- case HoodieTimeline.REPLACE_COMMIT_ACTION:
- case HoodieTimeline.CLUSTERING_ACTION:
- return commitDetail(record, "hoodieReplaceCommitMetadata",
skipMetadata);
- default: {
- throw new HoodieException("Unexpected action type: " + actionType);
+ private static String readArchivedMetadataString(HoodieArchivedTimeline
archivedTimeline, HoodieInstant instant,
+ boolean legacyArchive) {
+ Option<byte[]> details = archivedTimeline.getInstantDetails(instant);
+ if (!details.isPresent() || details.get().length == 0) {
+ // instants can be archived with no metadata, e.g. from an empty
completed
+ // meta file that a writer failure left behind
+ return "{}";
+ }
+ if (legacyArchive) {
+ // ArchivedTimelineV1 caches the JSON rendering of each archived entry,
which is what the
+ // legacy reader printed, and the v1 serde cannot decode it back into
the Avro classes
+ return new String(details.get(), StandardCharsets.UTF_8);
+ }
+ try {
+ // only the actions TimelineArchiverV2 archives can reach here, and it
archives completed
+ // compaction as commit, completed log compaction as deltacommit and
clustering as
+ // replacecommit; savepoints are never archived
+ switch (instant.getAction()) {
+ case HoodieTimeline.CLEAN_ACTION:
+ return archivedTimeline.readCleanMetadata(instant).toString();
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.DELTA_COMMIT_ACTION:
+ return
sortPartitions(archivedTimeline.readCommitMetadataToAvro(instant)).toString();
+ case HoodieTimeline.ROLLBACK_ACTION:
+ return archivedTimeline.readRollbackMetadata(instant).toString();
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ return
sortPartitions(archivedTimeline.readReplaceCommitMetadataToAvro(instant)).toString();
+ default:
+ throw new HoodieException("Unexpected action type: " +
instant.getAction());
Review Comment:
Fixed: the default arm decodes the payload through the schema embedded in
the Avro file, which is the rendering the typed records give of themselves, so
a savepoint, compaction plan or index metadata that the upgrade copied across,
or a legacy entry without an action state, lists like everything else instead
of failing the command. `{}` would hide metadata that is there.
Test: `testShowCommitsRendersActionsWithoutTypedReader` writes a completed
savepoint into the LSM timeline the way `upgradeToLSMTimeline` does
(`ActiveActionWithDetails` + `LSMTimelineWriter`) and lists it with metadata.
##########
hudi-cli/src/main/java/org/apache/hudi/cli/commands/RepairsCommand.java:
##########
@@ -189,29 +193,52 @@ 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.
+ */
+ private static boolean isReadableCleanerPlan(HoodieTableMetaClient client,
byte[] plan) {
+ try {
+ CleanerUtils.getCleanerPlan(client, new ByteArrayInputStream(plan));
+ return true;
+ } catch (IOException | AvroRuntimeException e) {
Review Comment:
Fixed: `IllegalArgumentException` is in the multi-catch. The half-length
fixture stays, since stopping inside the header is the read-failure case, and a
second fixture is a container opened with `DataFileWriter.create()` and closed
with no record, the case you describe; the test expects both to be removed
alongside the empty files.
--
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]
