private void flushData(boolean endInput) {
        final List<WriteStatus> writeStatus;
        if (this.writerHelper != null) {
            writeStatus = this.writerHelper.getWriteStatuses(this.taskID);
            this.currentInstant = this.writerHelper.getInstantTime();
        } else {
            writeStatus = Collections.emptyList();
            this.currentInstant = instantToWrite(false);
            LOG.info("No data to write in subtask [{}] for instant [{}]", taskID, this.currentInstant);
        }
        final WriteMetadataEvent event = WriteMetadataEvent.builder()
                .taskID(taskID)
                .instantTime(this.currentInstant)
                .writeStatus(writeStatus)
                .lastBatch(true)
                .endInput(endInput)
                .build();
        this.eventGateway.sendEventToCoordinator(event);
        // nullify the write helper for next ckp
        this.writerHelper = null;
        this.writeStatuses.addAll(writeStatus);
        // blocks flushing until the coordinator starts a new instant
        this.confirming = true;
    }
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