@SuppressWarnings("unchecked, rawtypes")
    private boolean flushBucket(DataBucket bucket) {
        String instant = instantToWrite(true);

        if (instant == null) {
            // in case there are empty checkpoints that has no input data
            LOG.info("No inflight instant when flushing data, skip.");
            return false;
        }

        List<HoodieRecord> records = bucket.writeBuffer();
        ValidationUtils.checkState(records.size() > 0, "Data bucket to flush has no buffering records");
        if (config.getBoolean(FlinkOptions.PRE_COMBINE)) {
            records = FlinkWriteHelper.newInstance().deduplicateRecords(records, (HoodieIndex) null, -1);
        }
        bucket.preWrite(records);
        final List<WriteStatus> writeStatus = new ArrayList<>(writeFunction.apply(records, instant));
        records.clear();
        final WriteMetadataEvent event = WriteMetadataEvent.builder()
                .taskID(taskID)
                .instantTime(instant) // the write instant may shift but the event still use the currentInstant.
                .writeStatus(writeStatus)
                .lastBatch(false)
                .endInput(false)
                .build();

        this.eventGateway.sendEventToCoordinator(event);
        writeStatuses.addAll(writeStatus);
        return 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