private static DataStream<HoodieRecord> streamBootstrap(
            Configuration conf,
            RowType rowType,
            DataStream<RowData> dataStream,
            boolean bounded) {
        DataStream<HoodieRecord> dataStream1 = rowDataToHoodieRecord(conf, rowType, dataStream);

        if (conf.getBoolean(FlinkOptions.INDEX_BOOTSTRAP_ENABLED) || bounded) {
            dataStream1 = dataStream1
                    .transform(
                            "index_bootstrap",
                            TypeInformation.of(HoodieRecord.class),
                            new BootstrapOperator<>(conf))
                    .setParallelism(
                            conf.getOptional(FlinkOptions.INDEX_BOOTSTRAP_TASKS).orElse(dataStream1.getParallelism()))
                    .uid(opUID("index_bootstrap", conf));
        }

        return dataStream1;
    }
Constructs bootstrap pipeline.
The bootstrap operator loads the existing data index (primary key to file id mapping),
then send the indexing data set to subsequent operator(usually the bucket assign operator).

@param conf       The configuration
@param rowType    The row type
@param dataStream The data stream
@param bounded    Whether the source is bounded
@param overwrite  Whether it is insert overwrite
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/utils/Pipelines.java