private void initWriterHelper() {
        final String instant = instantToWrite(true);
        if (instant == null) {
            // in case there are empty checkpoints that has no input data
            throw new HoodieException("No inflight instant when flushing data!");
        }
        this.writerHelper =
                new BulkInsertWriterHelper(this.config, this.writeClient.getHoodieTable(), this.writeClient.getConfig(),
                        instant, this.taskID, getRuntimeContext().getNumberOfParallelSubtasks(),
                        getRuntimeContext().getAttemptNumber(),
                        this.rowType);
    }
End input action for batch source.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/append/AppendWriteFunction.java