void offer(SinkBufferEntity bufferEntity) throws InterruptedException {
        if (!flushThreadAlive) {
            LOGGER.info(String.format("Flush thread already exit, ignore offer request for label[%s]",
                    bufferEntity.getLabel()));
            return;
        }

        long start = System.nanoTime();
        if (!flushQueue.offer(bufferEntity, sinkOptions.getSinkOfferTimeout(), TimeUnit.MILLISECONDS)) {
            throw new RuntimeException(
                    "Timeout while offering data to flushQueue, exceed " + sinkOptions.getSinkOfferTimeout()
                            + " ms, see " + StarRocksSinkOptions.SINK_BATCH_OFFER_TIMEOUT.key());
        }
        if (offerTimeNs != null) {
            offerTimeNs.update(System.nanoTime() - start);
        }
    }
async flush

@return false if met eof and flush thread will exit.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksSinkManager.java