@SuppressWarnings("unchecked, rawtypes")
    private void flushRemaining(boolean endInput) {
        this.currentInstant = instantToWrite(hasData());
        if (this.currentInstant == null) {
            // in case there are empty checkpoints that has no input data
            throw new HoodieException("No inflight instant when flushing data!");
        }
        final List<WriteStatus> writeStatus;
        if (buckets.size() > 0) {
            writeStatus = new ArrayList<>();
            this.buckets.values()
                    // The records are partitioned by the bucket ID and each batch sent to
                    // the writer belongs to one bucket.
                    .forEach(bucket -> {
                        List<HoodieRecord> records = bucket.writeBuffer();
                        if (records.size() > 0) {
                            if (config.getBoolean(FlinkOptions.PRE_COMBINE)) {
                                records = FlinkWriteHelper.newInstance().deduplicateRecords(records, (HoodieIndex) null,
                                        -1);
                            }
                            bucket.preWrite(records);
                            writeStatus.addAll(writeFunction.apply(records, currentInstant));
                            records.clear();
                            bucket.reset();
                        }
                    });
        } else {
            LOG.info("No data to write in subtask [{}] for instant [{}]", taskID, currentInstant);
            writeStatus = Collections.emptyList();
        }
        final WriteMetadataEvent event = WriteMetadataEvent.builder()
                .taskID(taskID)
                .instantTime(currentInstant)
                .writeStatus(writeStatus)
                .lastBatch(true)
                .endInput(endInput)
                .build();

        this.eventGateway.sendEventToCoordinator(event);
        this.buckets.clear();
        this.tracer.reset();
        this.writeClient.cleanHandles();
        this.writeStatuses.addAll(writeStatus);
        // blocks flushing until the coordinator starts a new instant
        this.confirming = true;
    }
Buffers the given record.

<p>Flush the data bucket first if the bucket records size is greater than
the configured value {@link FlinkOptions#WRITE_BATCH_SIZE}.

<p>Flush the max size data bucket if the total buffer size exceeds the configured
threshold {@link FlinkOptions#WRITE_TASK_MAX_SIZE}.

@param value HoodieRecord
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/StreamWriteFunction.java