protected void bufferRecord(HoodieRecord<?> value) {
        final String bucketID = getBucketID(value);

        DataBucket bucket = this.buckets.computeIfAbsent(bucketID,
                k -> new DataBucket(this.config.getDouble(FlinkOptions.WRITE_BATCH_SIZE), value));
        final DataItem item = DataItem.fromHoodieRecord(value);

        bucket.records.add(item);

        boolean flushBucket = bucket.detector.detect(item);
        boolean flushBuffer = this.tracer.trace(bucket.detector.lastRecordSize);
        if (flushBucket) {
            if (flushBucket(bucket)) {
                this.tracer.countDown(bucket.detector.totalSize);
                bucket.reset();
            }
        } else if (flushBuffer) {
            // find the max size bucket and flush it out
            DataBucket bucketToFlush = this.buckets.values().stream()
                    .max(Comparator.comparingLong(b -> b.detector.totalSize))
                    .orElseThrow(NoSuchElementException::new);
            if (flushBucket(bucketToFlush)) {
                this.tracer.countDown(bucketToFlush.detector.totalSize);
                bucketToFlush.reset();
            } else {
                LOG.warn("The buffer size hits the threshold {}, but still flush the max size data bucket failed!",
                        this.tracer.maxBufferSize);
            }
        }
    }
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