private boolean asyncFlush() throws Exception {
        SinkBufferEntity flushData = flushQueue.poll(FLUSH_QUEUE_POLL_TIMEOUT, TimeUnit.MILLISECONDS);
        if (flushData == null || (0 == flushData.getBatchCount() && !flushData.EOF())) {
            return true;
        }
        if (flushData.EOF()) {
            return false;
        }
        stopScheduler();

        String tableIdentifier = flushData.getDatabase() + "." + flushData.getTable();
        if (SchemaUpdateExceptionPolicy.STOP_PARTIAL == schemaUpdatePolicy && ignoreWriteTables.contains(
                tableIdentifier)) {
            LOGGER.warn(
                    String.format("Stop writing to db[%s] table[%s] because of former errors and stop_partial policy",
                            flushData.getDatabase(), flushData.getTable()));
            return true;
        }

        StarRocksSinkOptions.StreamLoadFormat format = sinkOptions.getStreamLoadFormat();
        LOGGER.info("StarRocks data format: {}", format.name());
        if (multipleSink && StarRocksSinkOptions.StreamLoadFormat.CSV.equals(format)) {
            sinkOptions.getSinkStreamLoadProperties().put("columns", flushData.getColumns());
        }

        LOGGER.info(String.format("Async stream load: db[%s] table[%s] rows[%d] bytes[%d] label[%s].",
                flushData.getDatabase(), flushData.getTable(), flushData.getBatchCount(), flushData.getBatchSize(),
                flushData.getLabel()));
        long startWithRetries = System.nanoTime();
        for (int i = 0; i <= sinkOptions.getSinkMaxRetries(); i++) {
            try {
                long start = System.nanoTime();
                // flush to StarRocks with stream load
                Map<String, Object> result = starrocksStreamLoadVisitor.doStreamLoad(flushData);
                LOGGER.info(String.format("Async stream load finished: label[%s].", flushData.getLabel()));
                // metrics
                if (null != totalFlushBytes) {
                    totalFlushBytes.inc(flushData.getBatchSize());
                    totalFlushRows.inc(flushData.getBatchCount());
                    totalFlushTime.inc(System.nanoTime() - startWithRetries);
                    totalFlushTimeWithoutRetries.inc(System.nanoTime() - start);
                    totalFlushSucceededTimes.inc();
                    flushTimeNs.update(System.nanoTime() - start);
                    updateMetricsFromStreamLoadResult(result);

                    if (null != metricData) {
                        if (multipleSink) {
                            metricData.outputMetrics(flushData.getDatabase(), flushData.getTable(),
                                    flushData.getBatchCount(), flushData.getBatchSize());
                        } else {
                            metricData.invoke(flushData.getBatchCount(), flushData.getBatchSize());
                        }
                    }
                }
                startScheduler();
                break;
            } catch (Exception e) {
                if (totalFlushFailedTimes != null) {
                    totalFlushFailedTimes.inc();
                }
                LOGGER.warn("Failed to flush batch data to StarRocks, retry times = {}", i, e);
                if (i >= sinkOptions.getSinkMaxRetries()) {
                    if (schemaUpdatePolicy == null
                            || SchemaUpdateExceptionPolicy.THROW_WITH_STOP == schemaUpdatePolicy) {
                        throw e;
                    } else if (SchemaUpdateExceptionPolicy.STOP_PARTIAL == schemaUpdatePolicy) {
                        ignoreWriteTables.add(tableIdentifier);
                    } else if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                        handleDirtyData(flushData, e);
                    }
                }
                if (e instanceof StarRocksStreamLoadFailedException
                        && ((StarRocksStreamLoadFailedException) e).needReCreateLabel()) {
                    String oldLabel = flushData.getLabel();
                    flushData.reGenerateLabel();
                    LOGGER.warn(String.format("Batch label changed from [%s] to [%s]", oldLabel, flushData.getLabel()));
                }
                try {
                    Thread.sleep(1000L * Math.min(i + 1, 10));
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    throw new IOException("Unable to flush, interrupted while doing another attempt", e);
                }
            }
        }
        return true;
    }
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