public final synchronized void writeRecords(SinkBufferEntity bufferEntity) throws IOException {
        checkFlushException();
        try {
            if (StarRocksSinkSemantic.EXACTLY_ONCE.equals(sinkOptions.getSemantic())) {
                return;
            }
            String database = bufferEntity.getDatabase();
            String table = bufferEntity.getTable();
            String bufferKey = String.format("%s,%s", database, table);
            if (bufferEntity.getBatchCount() >= sinkOptions.getSinkMaxRows()
                    || bufferEntity.getBatchSize() >= sinkOptions.getSinkMaxBytes()) {
                LOGGER.info(
                        String.format("StarRocks buffer Sinking triggered: db: [%s] table: [%s] rows[%d] label[%s].",
                                database, table, bufferEntity.getBatchCount(), bufferEntity.getLabel()));
                flush(bufferKey, false);
            }
        } catch (Exception e) {
            throw new IOException("Writing records to StarRocks failed.", e);
        }
    }
If a table writing throws exception, ignore it when receiving data later again
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksSinkManager.java