private static DataStream<HoodieRecord> boundedBootstrap(
            Configuration conf,
            RowType rowType,
            DataStream<RowData> dataStream) {
        final RowDataKeyGen rowDataKeyGen = RowDataKeyGen.instance(conf, rowType);
        // shuffle by partition keys
        dataStream = dataStream
                .keyBy(rowDataKeyGen::getPartitionPath);

        return rowDataToHoodieRecord(conf, rowType, dataStream)
                .transform(
                        "batch_index_bootstrap",
                        TypeInformation.of(HoodieRecord.class),
                        new BatchBootstrapOperator<>(conf))
                .setParallelism(
                        conf.getOptional(FlinkOptions.INDEX_BOOTSTRAP_TASKS).orElse(dataStream.getParallelism()))
                .uid(opUID("batch_index_bootstrap", conf));
    }
Constructs bootstrap pipeline for batch execution mode.
The indexing data set is loaded before the actual data write
in order to support batch UPSERT.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/utils/Pipelines.java