private void initWriteFunction() {
        final String writeOperation = this.config.get(FlinkOptions.OPERATION);
        switch (WriteOperationType.fromValue(writeOperation)) {
            case INSERT:
                this.writeFunction = (records, instantTime) -> this.writeClient.insert(records, instantTime);
                break;
            case UPSERT:
                this.writeFunction = (records, instantTime) -> this.writeClient.upsert(records, instantTime);
                break;
            case INSERT_OVERWRITE:
                this.writeFunction = (records, instantTime) -> this.writeClient.insertOverwrite(records, instantTime);
                break;
            case INSERT_OVERWRITE_TABLE:
                this.writeFunction =
                        (records, instantTime) -> this.writeClient.insertOverwriteTable(records, instantTime);
                break;
            default:
                throw new RuntimeException("Unsupported write operation : " + writeOperation);
        }
    }
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/StreamWriteFunction.java