@Override
  protected Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> buildMergeClusteringGroup(
      ConsistentBucketIdentifier identifier, List<FileSlice> fileSlices, int mergeSlot) {
    return super.buildMergeClusteringGroup(identifier, fileSlices, mergeSlot);
  }
@Test
  public void testBuildMergeClusteringGroup() throws Exception {
    setup();
    int maxFileSize = 5120;
    Properties props = new Properties();
    props.setProperty(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "uuid");
    HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(basePath)
        .withIndexConfig(HoodieIndexConfig.newBuilder().fromProperties(props).withIndexType(HoodieIndex.IndexType.BUCKET)
            .withBucketIndexEngineType(HoodieIndex.BucketIndexEngineType.CONSISTENT_HASHING)
            .withBucketMinNum(4)
            .withBucketNum("4").build())
        .withStorageConfig(HoodieStorageConfig.newBuilder()
            .parquetMaxFileSize(maxFileSize).build())
        .build();

    HoodieTable hoodieTable = HoodieSparkTable.create(config, context, metaClient);
    SparkConsistentBucketClusteringPlanStrategy planStrategy = new SparkConsistentBucketClusteringPlanStrategy(hoodieTable, context, config);

    HoodieConsistentHashingMetadata metadata = new HoodieConsistentHashingMetadata("partition", 8);
    ConsistentBucketIdentifier identifier = new ConsistentBucketIdentifier(metadata);

    int mergeSize = (int) (maxFileSize * BUCKET_MERGE_THRESHOLD.defaultValue());
    int[] fsSize = {0, maxFileSize, mergeSize / 2, mergeSize / 2 + 10, mergeSize / 2, maxFileSize, mergeSize / 4, mergeSize / 4};
    List<FileSlice> fileSlices = IntStream.range(0, metadata.getNodes().size()).mapToObj(
        i -> createFileSliceWithSize(metadata.getNodes().get(i).getFileIdPrefix(), fsSize[i] / 2, fsSize[i] / 2)
    ).collect(Collectors.toList());

    /**
     * 1. Test merge candidate selection based on file size
     * 2. Test empty file size
     * 3. Test merge slot
     */
    Triple res = planStrategy.buildMergeClusteringGroup(identifier, fileSlices, 4);
    Assertions.assertEquals(3, res.getMiddle());
    List<HoodieClusteringGroup> groups = (List<HoodieClusteringGroup>) res.getLeft();
    Assertions.assertEquals(2, groups.size());

    // Check group 0
    Assertions.assertEquals(fileSlices.get(0).getFileId(), groups.get(0).getSlices().get(2).getFileId());
    Assertions.assertEquals(fileSlices.get(7).getFileId(), groups.get(0).getSlices().get(1).getFileId());
    Assertions.assertEquals(fileSlices.get(6).getFileId(), groups.get(0).getSlices().get(0).getFileId());
    Assertions.assertEquals(3, groups.get(0).getSlices().size());
    List<ConsistentHashingNode> nodes = ConsistentHashingNode.fromJsonString(groups.get(0).getExtraMetadata().get(BaseConsistentHashingBucketClusteringPlanStrategy.METADATA_CHILD_NODE_KEY));
    Assertions.assertEquals(3, nodes.size());
    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, nodes.get(0).getTag());
    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, nodes.get(1).getTag());
    Assertions.assertEquals(ConsistentHashingNode.NodeTag.REPLACE, nodes.get(2).getTag());
    Assertions.assertEquals(metadata.getNodes().get(0).getValue(), nodes.get(2).getValue());

    // Check group 1
    Assertions.assertEquals(fileSlices.get(2).getFileId(), groups.get(1).getSlices().get(0).getFileId());
    Assertions.assertEquals(fileSlices.get(3).getFileId(), groups.get(1).getSlices().get(1).getFileId());
    Assertions.assertEquals(2, groups.get(1).getSlices().size());
    nodes = ConsistentHashingNode.fromJsonString(groups.get(1).getExtraMetadata().get(BaseConsistentHashingBucketClusteringPlanStrategy.METADATA_CHILD_NODE_KEY));
    Assertions.assertEquals(2, nodes.size());
    Assertions.assertEquals(ConsistentHashingNode.NodeTag.DELETE, nodes.get(0).getTag());
    Assertions.assertEquals(ConsistentHashingNode.NodeTag.REPLACE, nodes.get(1).getTag());
    Assertions.assertEquals(metadata.getNodes().get(3).getValue(), nodes.get(1).getValue());

    HoodieConsistentHashingMetadata metadata1 = new HoodieConsistentHashingMetadata("partition", 4);
    ConsistentBucketIdentifier identifier1 = new ConsistentBucketIdentifier(metadata1);

    int[] fsSize1 = {mergeSize / 4, mergeSize / 4, maxFileSize, mergeSize / 4};
    List<FileSlice> fileSlices1 = IntStream.range(0, metadata1.getNodes().size()).mapToObj(
        i -> createFileSliceWithSize(metadata1.getNodes().get(i).getFileIdPrefix(), fsSize1[i] / 2, fsSize1[i] / 2)
    ).collect(Collectors.toList());

    Triple<List<HoodieClusteringGroup>, Integer, List<FileSlice>> res1 = planStrategy.buildMergeClusteringGroup(identifier1,
        fileSlices1.stream().filter(fs -> fs.getTotalFileSize() < mergeSize).collect(Collectors.toList()), 4);
    Assertions.assertEquals(1, res1.getLeft().size(), "should have 1 clustering group");
    Assertions.assertEquals(3, res1.getLeft().get(0).getSlices().size(), "should have 3 input files");
  }