private void bootstrapIndexIfNeed(String partition) {
        if (bucketIndex.containsKey(partition)) {
            return;
        }
        LOG.info(String.format("Loading Hoodie Table %s, with path %s", this.metaClient.getTableConfig().getTableName(),
                this.metaClient.getBasePath() + "/" + partition));

        // Load existing fileID belongs to this task
        Map<Integer, String> bucketToFileIDMap = new HashMap<>();
        this.writeClient.getHoodieTable().getHoodieView().getLatestFileSlices(partition).forEach(fileSlice -> {
            String fileId = fileSlice.getFileId();
            int bucketNumber = BucketIdentifier.bucketIdFromFileId(fileId);
            if (isBucketToLoad(bucketNumber, partition)) {
                LOG.info(String.format("Should load this partition bucket %s with fileId %s", bucketNumber, fileId));
                // Validate that one bucketId has only ONE fileId
                if (bucketToFileIDMap.containsKey(bucketNumber)) {
                    throw new RuntimeException(String.format("Duplicate fileId %s from bucket %s of partition %s found "
                            + "during the BucketStreamWriteFunction index bootstrap.", fileId, bucketNumber,
                            partition));
                } else {
                    LOG.info(String.format("Adding fileId %s to the bucket %s of partition %s.", fileId, bucketNumber,
                            partition));
                    bucketToFileIDMap.put(bucketNumber, fileId);
                }
            }
        });
        bucketIndex.put(partition, bucketToFileIDMap);
    }
Get partition_bucket -> fileID mapping from the existing hudi table.
This is a required operation for each restart to avoid having duplicate file ids for one bucket.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/bucket/BucketStreamWriteFunction.java