hudi-client/src/test/java/org/apache/hudi/client/TestCompactionAdminClient.java - hudi - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hudi.client;

 import org.apache.hudi.client.CompactionAdminClient.ValidationOpResult;
 import org.apache.hudi.common.model.CompactionOperation;
 import org.apache.hudi.common.model.FileSlice;
 import org.apache.hudi.common.model.HoodieFileGroup;
 import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.model.HoodieTestUtils;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
 import org.apache.hudi.common.util.CompactionTestUtils;
 import org.apache.hudi.common.util.CompactionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.table.compact.OperationResult;

 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;

 import java.io.IOException;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;

 import static org.apache.hudi.client.CompactionAdminClient.getRenamingActionsToAlignWithCompactionOperation;
 import static org.apache.hudi.client.CompactionAdminClient.renameLogFile;
 import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;

 public class TestCompactionAdminClient extends TestHoodieClientBase {

   private static final Logger LOG = LogManager.getLogger(TestCompactionAdminClient.class);

   private HoodieTableMetaClient metaClient;
   private CompactionAdminClient client;

   @Before
   public void setUp() throws Exception {
     initPath();
     initSparkContexts();
     metaClient = HoodieTestUtils.init(HoodieTestUtils.getDefaultHadoopConf(), basePath, MERGE_ON_READ);
     client = new CompactionAdminClient(jsc, basePath);
   }

   @Test
   public void testUnscheduleCompactionPlan() throws Exception {
     int numEntriesPerInstant = 10;
     CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
         numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
     // THere are delta-commits after compaction instant
     validateUnSchedulePlan(client, "000", "001", numEntriesPerInstant, 2 * numEntriesPerInstant);
     // THere are delta-commits after compaction instant
     validateUnSchedulePlan(client, "002", "003", numEntriesPerInstant, 2 * numEntriesPerInstant);
     // THere are no delta-commits after compaction instant
     validateUnSchedulePlan(client, "004", "005", numEntriesPerInstant, 0);
     // THere are no delta-commits after compaction instant
     validateUnSchedulePlan(client, "006", "007", numEntriesPerInstant, 0);
   }

   @Test
   public void testUnscheduleCompactionFileId() throws Exception {
     int numEntriesPerInstant = 10;
     CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
         numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
     Map<String, CompactionOperation> instantsWithOp =
         Stream.of("001", "003", "005", "007").map(instant -> {
           try {
             return Pair.of(instant, CompactionUtils.getCompactionPlan(metaClient, instant));
           } catch (IOException ioe) {
             throw new HoodieException(ioe);
           }
         }).map(instantWithPlan -> instantWithPlan.getRight().getOperations().stream()
             .map(op -> Pair.of(instantWithPlan.getLeft(), CompactionOperation.convertFromAvroRecordInstance(op)))
             .findFirst().get()).collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
     // THere are delta-commits after compaction instant
     validateUnScheduleFileId(client, "000", "001", instantsWithOp.get("001"), 2);
     // THere are delta-commits after compaction instant
     validateUnScheduleFileId(client, "002", "003", instantsWithOp.get("003"), 2);
     // THere are no delta-commits after compaction instant
     validateUnScheduleFileId(client, "004", "005", instantsWithOp.get("005"), 0);
     // THere are no delta-commits after compaction instant
     validateUnScheduleFileId(client, "006", "007", instantsWithOp.get("007"), 0);
   }

   @Test
   public void testRepairCompactionPlan() throws Exception {
     int numEntriesPerInstant = 10;
     CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
         numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
     // THere are delta-commits after compaction instant
     validateRepair("000", "001", numEntriesPerInstant, 2 * numEntriesPerInstant);
     // THere are delta-commits after compaction instant
     validateRepair("002", "003", numEntriesPerInstant, 2 * numEntriesPerInstant);
     // THere are no delta-commits after compaction instant
     validateRepair("004", "005", numEntriesPerInstant, 0);
     // THere are no delta-commits after compaction instant
     validateRepair("006", "007", numEntriesPerInstant, 0);
   }

   private void validateRepair(String ingestionInstant, String compactionInstant, int numEntriesPerInstant,
       int expNumRepairs) throws Exception {
     List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles =
         validateUnSchedulePlan(client, ingestionInstant, compactionInstant, numEntriesPerInstant, expNumRepairs, true);
     metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
     List<ValidationOpResult> result = client.validateCompactionPlan(metaClient, compactionInstant, 1);
     if (expNumRepairs > 0) {
       Assert.assertTrue("Expect some failures in validation", result.stream().anyMatch(r -> !r.isSuccess()));
     }
     // Now repair
     List<Pair<HoodieLogFile, HoodieLogFile>> undoFiles =
         result.stream().flatMap(r -> getRenamingActionsToAlignWithCompactionOperation(metaClient,
             compactionInstant, r.getOperation(), Option.empty()).stream()).map(rn -> {
               try {
                 renameLogFile(metaClient, rn.getKey(), rn.getValue());
               } catch (IOException e) {
                 throw new HoodieIOException(e.getMessage(), e);
               }
               return rn;
             }).collect(Collectors.toList());
     Map<String, String> renameFilesFromUndo = undoFiles.stream()
         .collect(Collectors.toMap(p -> p.getRight().getPath().toString(), x -> x.getLeft().getPath().toString()));
     Map<String, String> expRenameFiles = renameFiles.stream()
         .collect(Collectors.toMap(p -> p.getLeft().getPath().toString(), x -> x.getRight().getPath().toString()));
     if (expNumRepairs > 0) {
       Assert.assertFalse("Rename Files must be non-empty", renameFiles.isEmpty());
     } else {
       Assert.assertTrue("Rename Files must be empty", renameFiles.isEmpty());
     }
     expRenameFiles.forEach((key, value) -> LOG.info("Key :" + key + " renamed to " + value + " rolled back to "
         + renameFilesFromUndo.get(key)));

     Assert.assertEquals("Undo must completely rollback renames", expRenameFiles, renameFilesFromUndo);
     // Now expect validation to succeed
     result = client.validateCompactionPlan(metaClient, compactionInstant, 1);
     Assert.assertTrue("Expect no failures in validation", result.stream().allMatch(OperationResult::isSuccess));
     Assert.assertEquals("Expected Num Repairs", expNumRepairs, undoFiles.size());
   }

   /**
    * Enssure compaction plan is valid.
    *
    * @param compactionInstant Compaction Instant
    */
   private void ensureValidCompactionPlan(String compactionInstant) throws Exception {
     metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
     // Ensure compaction-plan is good to begin with
     List<ValidationOpResult> validationResults = client.validateCompactionPlan(metaClient, compactionInstant, 1);
     Assert.assertFalse("Some validations failed",
         validationResults.stream().anyMatch(v -> !v.isSuccess()));
   }

   private void validateRenameFiles(List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles, String ingestionInstant,
       String compactionInstant, HoodieTableFileSystemView fsView) {
     // Ensure new names of log-files are on expected lines
     Set<HoodieLogFile> uniqNewLogFiles = new HashSet<>();
     Set<HoodieLogFile> uniqOldLogFiles = new HashSet<>();

     renameFiles.forEach(lfPair -> {
       Assert.assertFalse("Old Log File Names do not collide", uniqOldLogFiles.contains(lfPair.getKey()));
       Assert.assertFalse("New Log File Names do not collide", uniqNewLogFiles.contains(lfPair.getValue()));
       uniqOldLogFiles.add(lfPair.getKey());
       uniqNewLogFiles.add(lfPair.getValue());
     });

     renameFiles.forEach(lfPair -> {
       HoodieLogFile oldLogFile = lfPair.getLeft();
       HoodieLogFile newLogFile = lfPair.getValue();
       Assert.assertEquals("Base Commit time is expected", ingestionInstant, newLogFile.getBaseCommitTime());
       Assert.assertEquals("Base Commit time is expected", compactionInstant, oldLogFile.getBaseCommitTime());
       Assert.assertEquals("File Id is expected", oldLogFile.getFileId(), newLogFile.getFileId());
       HoodieLogFile lastLogFileBeforeCompaction =
           fsView.getLatestMergedFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], ingestionInstant)
               .filter(fs -> fs.getFileId().equals(oldLogFile.getFileId())).map(fs -> fs.getLogFiles().findFirst().get())
               .findFirst().get();
       Assert.assertEquals("Log Version expected",
           lastLogFileBeforeCompaction.getLogVersion() + oldLogFile.getLogVersion(), newLogFile.getLogVersion());
       Assert.assertTrue("Log version does not collide",
           newLogFile.getLogVersion() > lastLogFileBeforeCompaction.getLogVersion());
     });
   }

   /**
    * Validate Unschedule operations.
    */
   private List<Pair<HoodieLogFile, HoodieLogFile>> validateUnSchedulePlan(CompactionAdminClient client,
       String ingestionInstant, String compactionInstant, int numEntriesPerInstant, int expNumRenames) throws Exception {
     return validateUnSchedulePlan(client, ingestionInstant, compactionInstant, numEntriesPerInstant, expNumRenames,
         false);
   }

   /**
    * Validate Unschedule operations.
    */
   private List<Pair<HoodieLogFile, HoodieLogFile>> validateUnSchedulePlan(CompactionAdminClient client,
       String ingestionInstant, String compactionInstant, int numEntriesPerInstant, int expNumRenames,
       boolean skipUnSchedule) throws Exception {

     ensureValidCompactionPlan(compactionInstant);

     // Check suggested rename operations
     List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles =
         client.getRenamingActionsForUnschedulingCompactionPlan(metaClient, compactionInstant, 1, Option.empty(), false);
     metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);

     // Log files belonging to file-slices created because of compaction request must be renamed

     Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(Pair::getLeft).collect(Collectors.toSet());
     final HoodieTableFileSystemView fsView =
         new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
     Set<HoodieLogFile> expLogFilesToBeRenamed = fsView.getLatestFileSlices(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0])
         .filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).flatMap(FileSlice::getLogFiles)
         .collect(Collectors.toSet());
     Assert.assertEquals("Log files belonging to file-slices created because of compaction request must be renamed",
         expLogFilesToBeRenamed, gotLogFilesToBeRenamed);

     if (skipUnSchedule) {
       // Do the renaming only but do not touch the compaction plan - Needed for repair tests
       renameFiles.forEach(lfPair -> {
         try {
           renameLogFile(metaClient, lfPair.getLeft(), lfPair.getRight());
         } catch (IOException e) {
           throw new HoodieIOException(e.getMessage(), e);
         }
       });
     } else {
       validateRenameFiles(renameFiles, ingestionInstant, compactionInstant, fsView);
     }

     Map<String, Long> fileIdToCountsBeforeRenaming =
         fsView.getLatestMergedFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant)
             .filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
             .map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
             .collect(Collectors.toMap(Pair::getKey, Pair::getValue));

     // Call the main unschedule API

     client.unscheduleCompactionPlan(compactionInstant, false, 1, false);

     metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
     final HoodieTableFileSystemView newFsView =
         new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
     // Expect all file-slice whose base-commit is same as compaction commit to contain no new Log files
     newFsView.getLatestFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant, true)
             .filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).forEach(fs -> {
               Assert.assertFalse("No Data file must be present", fs.getBaseFile().isPresent());
               Assert.assertEquals("No Log Files", 0, fs.getLogFiles().count());
             });

     // Ensure same number of log-files before and after renaming per fileId
     Map<String, Long> fileIdToCountsAfterRenaming =
         newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(HoodieFileGroup::getAllFileSlices)
             .filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
             .map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
             .collect(Collectors.toMap(Pair::getKey, Pair::getValue));

     Assert.assertEquals("Each File Id has same number of log-files", fileIdToCountsBeforeRenaming,
         fileIdToCountsAfterRenaming);
     Assert.assertEquals("Not Empty", numEntriesPerInstant, fileIdToCountsAfterRenaming.size());
     Assert.assertEquals("Expected number of renames", expNumRenames, renameFiles.size());
     return renameFiles;
   }

   /**
    * Validate Unschedule operations.
    */
   private void validateUnScheduleFileId(CompactionAdminClient client, String ingestionInstant, String compactionInstant,
       CompactionOperation op, int expNumRenames) throws Exception {

     ensureValidCompactionPlan(compactionInstant);

     // Check suggested rename operations
     List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles = client
         .getRenamingActionsForUnschedulingCompactionOperation(metaClient, compactionInstant, op, Option.empty(), false);
     metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);

     // Log files belonging to file-slices created because of compaction request must be renamed

     Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(Pair::getLeft).collect(Collectors.toSet());
     final HoodieTableFileSystemView fsView =
         new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
     Set<HoodieLogFile> expLogFilesToBeRenamed = fsView.getLatestFileSlices(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0])
         .filter(fs -> fs.getBaseInstantTime().equals(compactionInstant))
         .filter(fs -> fs.getFileId().equals(op.getFileId())).flatMap(FileSlice::getLogFiles)
         .collect(Collectors.toSet());
     Assert.assertEquals("Log files belonging to file-slices created because of compaction request must be renamed",
         expLogFilesToBeRenamed, gotLogFilesToBeRenamed);
     validateRenameFiles(renameFiles, ingestionInstant, compactionInstant, fsView);

     Map<String, Long> fileIdToCountsBeforeRenaming =
         fsView.getLatestMergedFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant)
             .filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
             .filter(fs -> fs.getFileId().equals(op.getFileId()))
             .map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
             .collect(Collectors.toMap(Pair::getKey, Pair::getValue));

     // Call the main unschedule API
     client.unscheduleCompactionFileId(op.getFileGroupId(), false, false);

     metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
     final HoodieTableFileSystemView newFsView =
         new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
     // Expect all file-slice whose base-commit is same as compaction commit to contain no new Log files
     newFsView.getLatestFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant, true)
         .filter(fs -> fs.getBaseInstantTime().equals(compactionInstant))
         .filter(fs -> fs.getFileId().equals(op.getFileId())).forEach(fs -> {
           Assert.assertFalse("No Data file must be present", fs.getBaseFile().isPresent());
           Assert.assertEquals("No Log Files", 0, fs.getLogFiles().count());
         });

     // Ensure same number of log-files before and after renaming per fileId
     Map<String, Long> fileIdToCountsAfterRenaming =
         newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(HoodieFileGroup::getAllFileSlices)
             .filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
             .filter(fs -> fs.getFileId().equals(op.getFileId()))
             .map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
             .collect(Collectors.toMap(Pair::getKey, Pair::getValue));

     Assert.assertEquals("Each File Id has same number of log-files", fileIdToCountsBeforeRenaming,
         fileIdToCountsAfterRenaming);
     Assert.assertEquals("Not Empty", 1, fileIdToCountsAfterRenaming.size());
     Assert.assertEquals("Expected number of renames", expNumRenames, renameFiles.size());
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hudi.client;

	import org.apache.hudi.client.CompactionAdminClient.ValidationOpResult;
	import org.apache.hudi.common.model.CompactionOperation;
	import org.apache.hudi.common.model.FileSlice;
	import org.apache.hudi.common.model.HoodieFileGroup;
	import org.apache.hudi.common.model.HoodieLogFile;
	import org.apache.hudi.common.model.HoodieTestUtils;
	import org.apache.hudi.common.table.HoodieTableMetaClient;
	import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
	import org.apache.hudi.common.util.CompactionTestUtils;
	import org.apache.hudi.common.util.CompactionUtils;
	import org.apache.hudi.common.util.Option;
	import org.apache.hudi.common.util.collection.Pair;
	import org.apache.hudi.exception.HoodieException;
	import org.apache.hudi.exception.HoodieIOException;
	import org.apache.hudi.table.compact.OperationResult;

	import org.apache.log4j.LogManager;
	import org.apache.log4j.Logger;
	import org.junit.Assert;
	import org.junit.Before;
	import org.junit.Test;

	import java.io.IOException;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.stream.Collectors;
	import java.util.stream.Stream;

	import static org.apache.hudi.client.CompactionAdminClient.getRenamingActionsToAlignWithCompactionOperation;
	import static org.apache.hudi.client.CompactionAdminClient.renameLogFile;
	import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;

	public class TestCompactionAdminClient extends TestHoodieClientBase {

	private static final Logger LOG = LogManager.getLogger(TestCompactionAdminClient.class);

	private HoodieTableMetaClient metaClient;
	private CompactionAdminClient client;

	@Before
	public void setUp() throws Exception {
	initPath();
	initSparkContexts();
	metaClient = HoodieTestUtils.init(HoodieTestUtils.getDefaultHadoopConf(), basePath, MERGE_ON_READ);
	client = new CompactionAdminClient(jsc, basePath);
	}

	@Test
	public void testUnscheduleCompactionPlan() throws Exception {
	int numEntriesPerInstant = 10;
	CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
	numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
	// THere are delta-commits after compaction instant
	validateUnSchedulePlan(client, "000", "001", numEntriesPerInstant, 2 * numEntriesPerInstant);
	// THere are delta-commits after compaction instant
	validateUnSchedulePlan(client, "002", "003", numEntriesPerInstant, 2 * numEntriesPerInstant);
	// THere are no delta-commits after compaction instant
	validateUnSchedulePlan(client, "004", "005", numEntriesPerInstant, 0);
	// THere are no delta-commits after compaction instant
	validateUnSchedulePlan(client, "006", "007", numEntriesPerInstant, 0);
	}

	@Test
	public void testUnscheduleCompactionFileId() throws Exception {
	int numEntriesPerInstant = 10;
	CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
	numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
	Map<String, CompactionOperation> instantsWithOp =
	Stream.of("001", "003", "005", "007").map(instant -> {
	try {
	return Pair.of(instant, CompactionUtils.getCompactionPlan(metaClient, instant));
	} catch (IOException ioe) {
	throw new HoodieException(ioe);
	}
	}).map(instantWithPlan -> instantWithPlan.getRight().getOperations().stream()
	.map(op -> Pair.of(instantWithPlan.getLeft(), CompactionOperation.convertFromAvroRecordInstance(op)))
	.findFirst().get()).collect(Collectors.toMap(Pair::getLeft, Pair::getRight));
	// THere are delta-commits after compaction instant
	validateUnScheduleFileId(client, "000", "001", instantsWithOp.get("001"), 2);
	// THere are delta-commits after compaction instant
	validateUnScheduleFileId(client, "002", "003", instantsWithOp.get("003"), 2);
	// THere are no delta-commits after compaction instant
	validateUnScheduleFileId(client, "004", "005", instantsWithOp.get("005"), 0);
	// THere are no delta-commits after compaction instant
	validateUnScheduleFileId(client, "006", "007", instantsWithOp.get("007"), 0);
	}

	@Test
	public void testRepairCompactionPlan() throws Exception {
	int numEntriesPerInstant = 10;
	CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
	numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
	// THere are delta-commits after compaction instant
	validateRepair("000", "001", numEntriesPerInstant, 2 * numEntriesPerInstant);
	// THere are delta-commits after compaction instant
	validateRepair("002", "003", numEntriesPerInstant, 2 * numEntriesPerInstant);
	// THere are no delta-commits after compaction instant
	validateRepair("004", "005", numEntriesPerInstant, 0);
	// THere are no delta-commits after compaction instant
	validateRepair("006", "007", numEntriesPerInstant, 0);
	}

	private void validateRepair(String ingestionInstant, String compactionInstant, int numEntriesPerInstant,
	int expNumRepairs) throws Exception {
	List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles =
	validateUnSchedulePlan(client, ingestionInstant, compactionInstant, numEntriesPerInstant, expNumRepairs, true);
	metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
	List<ValidationOpResult> result = client.validateCompactionPlan(metaClient, compactionInstant, 1);
	if (expNumRepairs > 0) {
	Assert.assertTrue("Expect some failures in validation", result.stream().anyMatch(r -> !r.isSuccess()));
	}
	// Now repair
	List<Pair<HoodieLogFile, HoodieLogFile>> undoFiles =
	result.stream().flatMap(r -> getRenamingActionsToAlignWithCompactionOperation(metaClient,
	compactionInstant, r.getOperation(), Option.empty()).stream()).map(rn -> {
	try {
	renameLogFile(metaClient, rn.getKey(), rn.getValue());
	} catch (IOException e) {
	throw new HoodieIOException(e.getMessage(), e);
	}
	return rn;
	}).collect(Collectors.toList());
	Map<String, String> renameFilesFromUndo = undoFiles.stream()
	.collect(Collectors.toMap(p -> p.getRight().getPath().toString(), x -> x.getLeft().getPath().toString()));
	Map<String, String> expRenameFiles = renameFiles.stream()
	.collect(Collectors.toMap(p -> p.getLeft().getPath().toString(), x -> x.getRight().getPath().toString()));
	if (expNumRepairs > 0) {
	Assert.assertFalse("Rename Files must be non-empty", renameFiles.isEmpty());
	} else {
	Assert.assertTrue("Rename Files must be empty", renameFiles.isEmpty());
	}
	expRenameFiles.forEach((key, value) -> LOG.info("Key :" + key + " renamed to " + value + " rolled back to "
	+ renameFilesFromUndo.get(key)));

	Assert.assertEquals("Undo must completely rollback renames", expRenameFiles, renameFilesFromUndo);
	// Now expect validation to succeed
	result = client.validateCompactionPlan(metaClient, compactionInstant, 1);
	Assert.assertTrue("Expect no failures in validation", result.stream().allMatch(OperationResult::isSuccess));
	Assert.assertEquals("Expected Num Repairs", expNumRepairs, undoFiles.size());
	}

	/**
	* Enssure compaction plan is valid.
	*
	* @param compactionInstant Compaction Instant
	*/
	private void ensureValidCompactionPlan(String compactionInstant) throws Exception {
	metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
	// Ensure compaction-plan is good to begin with
	List<ValidationOpResult> validationResults = client.validateCompactionPlan(metaClient, compactionInstant, 1);
	Assert.assertFalse("Some validations failed",
	validationResults.stream().anyMatch(v -> !v.isSuccess()));
	}

	private void validateRenameFiles(List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles, String ingestionInstant,
	String compactionInstant, HoodieTableFileSystemView fsView) {
	// Ensure new names of log-files are on expected lines
	Set<HoodieLogFile> uniqNewLogFiles = new HashSet<>();
	Set<HoodieLogFile> uniqOldLogFiles = new HashSet<>();

	renameFiles.forEach(lfPair -> {
	Assert.assertFalse("Old Log File Names do not collide", uniqOldLogFiles.contains(lfPair.getKey()));
	Assert.assertFalse("New Log File Names do not collide", uniqNewLogFiles.contains(lfPair.getValue()));
	uniqOldLogFiles.add(lfPair.getKey());
	uniqNewLogFiles.add(lfPair.getValue());
	});

	renameFiles.forEach(lfPair -> {
	HoodieLogFile oldLogFile = lfPair.getLeft();
	HoodieLogFile newLogFile = lfPair.getValue();
	Assert.assertEquals("Base Commit time is expected", ingestionInstant, newLogFile.getBaseCommitTime());
	Assert.assertEquals("Base Commit time is expected", compactionInstant, oldLogFile.getBaseCommitTime());
	Assert.assertEquals("File Id is expected", oldLogFile.getFileId(), newLogFile.getFileId());
	HoodieLogFile lastLogFileBeforeCompaction =
	fsView.getLatestMergedFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], ingestionInstant)
	.filter(fs -> fs.getFileId().equals(oldLogFile.getFileId())).map(fs -> fs.getLogFiles().findFirst().get())
	.findFirst().get();
	Assert.assertEquals("Log Version expected",
	lastLogFileBeforeCompaction.getLogVersion() + oldLogFile.getLogVersion(), newLogFile.getLogVersion());
	Assert.assertTrue("Log version does not collide",
	newLogFile.getLogVersion() > lastLogFileBeforeCompaction.getLogVersion());
	});
	}

	/**
	* Validate Unschedule operations.
	*/
	private List<Pair<HoodieLogFile, HoodieLogFile>> validateUnSchedulePlan(CompactionAdminClient client,
	String ingestionInstant, String compactionInstant, int numEntriesPerInstant, int expNumRenames) throws Exception {
	return validateUnSchedulePlan(client, ingestionInstant, compactionInstant, numEntriesPerInstant, expNumRenames,
	false);
	}

	/**
	* Validate Unschedule operations.
	*/
	private List<Pair<HoodieLogFile, HoodieLogFile>> validateUnSchedulePlan(CompactionAdminClient client,
	String ingestionInstant, String compactionInstant, int numEntriesPerInstant, int expNumRenames,
	boolean skipUnSchedule) throws Exception {

	ensureValidCompactionPlan(compactionInstant);

	// Check suggested rename operations
	List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles =
	client.getRenamingActionsForUnschedulingCompactionPlan(metaClient, compactionInstant, 1, Option.empty(), false);
	metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);

	// Log files belonging to file-slices created because of compaction request must be renamed

	Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(Pair::getLeft).collect(Collectors.toSet());
	final HoodieTableFileSystemView fsView =
	new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
	Set<HoodieLogFile> expLogFilesToBeRenamed = fsView.getLatestFileSlices(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0])
	.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).flatMap(FileSlice::getLogFiles)
	.collect(Collectors.toSet());
	Assert.assertEquals("Log files belonging to file-slices created because of compaction request must be renamed",
	expLogFilesToBeRenamed, gotLogFilesToBeRenamed);

	if (skipUnSchedule) {
	// Do the renaming only but do not touch the compaction plan - Needed for repair tests
	renameFiles.forEach(lfPair -> {
	try {
	renameLogFile(metaClient, lfPair.getLeft(), lfPair.getRight());
	} catch (IOException e) {
	throw new HoodieIOException(e.getMessage(), e);
	}
	});
	} else {
	validateRenameFiles(renameFiles, ingestionInstant, compactionInstant, fsView);
	}

	Map<String, Long> fileIdToCountsBeforeRenaming =
	fsView.getLatestMergedFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant)
	.filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
	.map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
	.collect(Collectors.toMap(Pair::getKey, Pair::getValue));

	// Call the main unschedule API

	client.unscheduleCompactionPlan(compactionInstant, false, 1, false);

	metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
	final HoodieTableFileSystemView newFsView =
	new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
	// Expect all file-slice whose base-commit is same as compaction commit to contain no new Log files
	newFsView.getLatestFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant, true)
	.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).forEach(fs -> {
	Assert.assertFalse("No Data file must be present", fs.getBaseFile().isPresent());
	Assert.assertEquals("No Log Files", 0, fs.getLogFiles().count());
	});

	// Ensure same number of log-files before and after renaming per fileId
	Map<String, Long> fileIdToCountsAfterRenaming =
	newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(HoodieFileGroup::getAllFileSlices)
	.filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
	.map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
	.collect(Collectors.toMap(Pair::getKey, Pair::getValue));

	Assert.assertEquals("Each File Id has same number of log-files", fileIdToCountsBeforeRenaming,
	fileIdToCountsAfterRenaming);
	Assert.assertEquals("Not Empty", numEntriesPerInstant, fileIdToCountsAfterRenaming.size());
	Assert.assertEquals("Expected number of renames", expNumRenames, renameFiles.size());
	return renameFiles;
	}

	/**
	* Validate Unschedule operations.
	*/
	private void validateUnScheduleFileId(CompactionAdminClient client, String ingestionInstant, String compactionInstant,
	CompactionOperation op, int expNumRenames) throws Exception {

	ensureValidCompactionPlan(compactionInstant);

	// Check suggested rename operations
	List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles = client
	.getRenamingActionsForUnschedulingCompactionOperation(metaClient, compactionInstant, op, Option.empty(), false);
	metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);

	// Log files belonging to file-slices created because of compaction request must be renamed

	Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(Pair::getLeft).collect(Collectors.toSet());
	final HoodieTableFileSystemView fsView =
	new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
	Set<HoodieLogFile> expLogFilesToBeRenamed = fsView.getLatestFileSlices(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0])
	.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant))
	.filter(fs -> fs.getFileId().equals(op.getFileId())).flatMap(FileSlice::getLogFiles)
	.collect(Collectors.toSet());
	Assert.assertEquals("Log files belonging to file-slices created because of compaction request must be renamed",
	expLogFilesToBeRenamed, gotLogFilesToBeRenamed);
	validateRenameFiles(renameFiles, ingestionInstant, compactionInstant, fsView);

	Map<String, Long> fileIdToCountsBeforeRenaming =
	fsView.getLatestMergedFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant)
	.filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
	.filter(fs -> fs.getFileId().equals(op.getFileId()))
	.map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
	.collect(Collectors.toMap(Pair::getKey, Pair::getValue));

	// Call the main unschedule API
	client.unscheduleCompactionFileId(op.getFileGroupId(), false, false);

	metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
	final HoodieTableFileSystemView newFsView =
	new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
	// Expect all file-slice whose base-commit is same as compaction commit to contain no new Log files
	newFsView.getLatestFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant, true)
	.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant))
	.filter(fs -> fs.getFileId().equals(op.getFileId())).forEach(fs -> {
	Assert.assertFalse("No Data file must be present", fs.getBaseFile().isPresent());
	Assert.assertEquals("No Log Files", 0, fs.getLogFiles().count());
	});

	// Ensure same number of log-files before and after renaming per fileId
	Map<String, Long> fileIdToCountsAfterRenaming =
	newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(HoodieFileGroup::getAllFileSlices)
	.filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
	.filter(fs -> fs.getFileId().equals(op.getFileId()))
	.map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
	.collect(Collectors.toMap(Pair::getKey, Pair::getValue));

	Assert.assertEquals("Each File Id has same number of log-files", fileIdToCountsBeforeRenaming,
	fileIdToCountsAfterRenaming);
	Assert.assertEquals("Not Empty", 1, fileIdToCountsAfterRenaming.size());
	Assert.assertEquals("Expected number of renames", expNumRenames, renameFiles.size());
	}
	}