public void endInput() {
        super.endInput();
        flushRemaining(true);
        this.writeClient.cleanHandles();
        this.writeStatuses.clear();
    }
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