voonhous commented on code in PR #18816:
URL: https://github.com/apache/hudi/pull/18816#discussion_r3956841501
##########
hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestRepairsCommand.java:
##########
@@ -256,49 +279,220 @@ public void testRemoveCorruptedPendingCleanAction()
throws IOException {
// Create four requested files
for (int i = 100; i < 104; i++) {
String timestamp = String.valueOf(i);
- // Write corrupted requested Clean File
+ // Write an empty requested Clean File
HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
timestamp, conf);
}
+ // A plan cut short mid-write, so that its bytes stop inside the Avro
header
+ FileCreateUtils.createRequestedCleanFile(metaClient, "104",
validCleanerPlan());
+ truncateInHalf(metaClient, requestedCleanPath(metaClient, "104"));
+
+ // A plan whose writer was killed between opening and closing the Avro
container, which leaves a
+ // complete header and no record behind
+ try (DataFileWriter<HoodieCleanerPlan> writer =
+ new DataFileWriter<>(new
SpecificDatumWriter<>(HoodieCleanerPlan.class))) {
+ writer.create(HoodieCleanerPlan.getClassSchema(),
+ metaClient.getStorage().create(requestedCleanPath(metaClient,
"105"), true));
+ }
+
+ // A plan that decodes, which the command has to leave in place
+ FileCreateUtils.createRequestedCleanFile(metaClient, "106",
validCleanerPlan());
+
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- // first, there are four instants
- assertEquals(4,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+ // first, there are seven pending instants
+ assertEquals(7,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
- Object result = shell.evaluate(() -> "repair corrupted clean files");
- assertTrue(ShellEvaluationResultUtil.isSuccess(result));
+ Object cleanResult = shell.evaluate(() -> "repair corrupted clean files");
+ assertTrue(ShellEvaluationResultUtil.isSuccess(cleanResult));
// reload meta client
metaClient = HoodieTableMetaClient.reload(metaClient);
- assertEquals(0,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+ // the empty, the truncated and the record-less plans are gone and the
readable one is untouched
+ List<HoodieInstant> remaining =
+
metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, remaining.size());
+ assertEquals("106", remaining.get(0).requestedTime());
+ }
+
+ /**
+ * A transient read failure on a valid pending clean plan must not be taken
for corruption:
+ * the timeline serde wraps any exception raised while it streams the plan
file in the same
+ * "unable to read commit metadata" IOException that an empty or truncated
file raises.
+ */
+ @Test
+ public void testRemoveCorruptedPendingCleanActionKeepsPlanOnReadFailure()
throws IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+ FileCreateUtils.createRequestedCleanFile(metaClient, "100",
validCleanerPlan());
+ StoragePath planPath = requestedCleanPath(metaClient, "100");
+
+ HoodieTableMetaClient timingOutClient = HoodieTableMetaClient.builder()
+ .setStorage(new TimingOutStorage(fs,
planPath)).setBasePath(tablePath).build();
+ HoodieIOException thrown = assertThrows(HoodieIOException.class,
+ () ->
RepairsCommand.removeCorruptedPendingCleanAction(timingOutClient));
+ assertInstanceOf(SocketTimeoutException.class, thrown.getCause());
+ assertTrue(metaClient.getStorage().exists(planPath));
+
+ // the same plan read through a healthy storage is left alone as well
+
RepairsCommand.removeCorruptedPendingCleanAction(HoodieTableMetaClient.reload(metaClient));
+ assertEquals(1,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterInflightsAndRequested().countInstants());
+ }
+
+ /**
+ * A clean that reached inflight keeps its plan in the requested file, so a
corrupt plan has to
+ * take both files with it. The timeline collapses the two states into the
inflight instant
+ * alone, so removing only the instant it listed would leave the corrupt
plan behind for the
+ * next clean to fail on, and take a second run of this command to clear.
+ */
+ @Test
+ public void testRemoveCorruptedPendingCleanActionRemovesInflightAndItsPlan()
throws IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ // a clean that was scheduled and started, whose plan was then cut short
+ FileCreateUtils.createRequestedCleanFile(metaClient, "100",
validCleanerPlan());
+ // the inflight file a clean leaves behind carries no plan of its own
+ FileCreateUtils.createInflightCleanFile(metaClient, "100", null, true);
+ truncateInHalf(metaClient, requestedCleanPath(metaClient, "100"));
+
+ StoragePath inflightPath = new StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeInflightCleanerFileName("100"));
+ assertTrue(metaClient.getStorage().exists(inflightPath));
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ List<HoodieInstant> pending =
metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, pending.size());
+ assertTrue(pending.get(0).isInflight());
+
+ RepairsCommand.removeCorruptedPendingCleanAction(metaClient);
+
+ // one pass takes the whole action, not just the file the timeline listed
+ assertFalse(metaClient.getStorage().exists(inflightPath));
+ assertFalse(metaClient.getStorage().exists(requestedCleanPath(metaClient,
"100")));
+ assertEquals(0,
HoodieTableMetaClient.reload(metaClient).getActiveTimeline()
+ .filterInflightsAndRequested().countInstants());
+ }
+
+ /**
+ * Corrupt bytes do not always reach the decoder as an Avro failure: a plan
that decodes with no
+ * version leaves the migrator to unbox a null, and a length that survives
as far as Avro's own
+ * ceiling raises an {@code UnsupportedOperationException}. Such an instant
has to be judged
+ * corrupt like any other, and the instants behind it still repaired, rather
than the failure
+ * escaping and abandoning the rest of the timeline.
+ */
+ @Test
+ public void
testRemoveCorruptedPendingCleanActionRepairsPastAnUndecodablePlan() throws
IOException {
+ HoodieCLI.conf = storageConf();
+ HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+ HoodieCleanerPlan versionless = validCleanerPlan();
+ versionless.setVersion(null);
+ writeCleanerPlan(metaClient, "100", versionless);
+
+ // an ordinary corruption behind it, which is only reached if the first
one does not escape
+ HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath,
"101", HoodieCLI.conf);
+ // and a readable plan the command has to leave in place
+ FileCreateUtils.createRequestedCleanFile(metaClient, "102",
validCleanerPlan());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertEquals(3,
metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
+
+ RepairsCommand.removeCorruptedPendingCleanAction(metaClient);
+
+ List<HoodieInstant> remaining = HoodieTableMetaClient.reload(metaClient)
+ .getActiveTimeline().filterInflightsAndRequested().getInstants();
+ assertEquals(1, remaining.size());
+ assertEquals("102", remaining.get(0).requestedTime());
+ }
+
+ /**
+ * Cuts a file's bytes in half in place. The file is overwritten rather than
deleted and
+ * rewritten, because both the emptiness check and the file name generator
resolve the file
+ * from the instant itself.
+ */
+ private static void truncateInHalf(HoodieTableMetaClient metaClient,
StoragePath path) throws IOException {
+ byte[] bytes;
+ try (InputStream in = metaClient.getStorage().open(path)) {
+ bytes = FileIOUtils.readAsByteArray(in);
+ }
+ try (OutputStream out = metaClient.getStorage().create(path, true)) {
+ out.write(bytes, 0, bytes.length / 2);
+ }
+ }
+
+ /**
+ * Writes a clean plan straight into the requested file of an instant, so
that a plan the
+ * timeline's own writer would not produce still reaches the command.
+ */
+ private static void writeCleanerPlan(HoodieTableMetaClient metaClient,
String instantTime,
+ HoodieCleanerPlan plan) throws
IOException {
+ try (DataFileWriter<HoodieCleanerPlan> writer =
+ new DataFileWriter<>(new
SpecificDatumWriter<>(HoodieCleanerPlan.class))) {
+ writer.create(HoodieCleanerPlan.getClassSchema(),
+ metaClient.getStorage().create(requestedCleanPath(metaClient,
instantTime), true));
+ writer.append(plan);
+ }
+ }
+
+ /**
+ * The path of the requested file of a clean instant.
+ */
+ private static StoragePath requestedCleanPath(HoodieTableMetaClient
metaClient, String instantTime) {
+ return new StoragePath(metaClient.getTimelinePath(),
+
metaClient.getInstantFileNameGenerator().makeRequestedCleanerFileName(instantTime));
+ }
+
+ /**
+ * A clean plan that decodes into the latest plan version.
+ */
+ private static HoodieCleanerPlan validCleanerPlan() {
+ return HoodieCleanerPlan.newBuilder()
+ .setEarliestInstantToRetain(HoodieActionInstant.newBuilder()
+ .setAction(HoodieTimeline.COMMIT_ACTION).setTimestamp("001")
+ .setState(HoodieInstant.State.COMPLETED.name()).build())
+
.setFilesToBeDeletedPerPartition(Collections.singletonMap("partition1",
Collections.singletonList("file1")))
+
.setFilePathsToBeDeletedPerPartition(Collections.singletonMap("partition1",
+
Collections.singletonList(HoodieCleanFileInfo.newBuilder().setFilePath("file1").build())))
+ .setLastCompletedCommitTimestamp("002")
+ .setPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name())
+ .setVersion(TimelineLayoutVersion.CURR_VERSION)
Review Comment:
Right -- the two are both `2` by coincidence.
`TimelineLayoutVersion.CURR_VERSION` is the timeline layout, while
`CleanPlanMigrator` dispatches on the clean plan's own schema version. Switched
to `CleanPlanV2MigrationHandler.VERSION`, matching `addClean`.
--
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]
