public final synchronized void writeRecords(String database, String table, String... records) throws IOException {
        checkFlushException();
        try {
            if (0 == records.length) {
                return;
            }
            String bufferKey = String.format("%s,%s", database, table);
            SinkBufferEntity bufferEntity = bufferMap.computeIfAbsent(bufferKey,
                    k -> new SinkBufferEntity(database, table, sinkOptions.getLabelPrefix()));
            bufferEntity.setDirtyLogTag(dirtySinkHelper.getDirtyOptions().getLogTag());
            bufferEntity.setDirtyIdentify(dirtySinkHelper.getDirtyOptions().getIdentifier());
            bufferEntity.setDirtyLabel(dirtySinkHelper.getDirtyOptions().getLabels());

            for (String record : records) {
                byte[] bts = record.getBytes(StandardCharsets.UTF_8);
                bufferEntity.addToBuffer(bts);
            }
            writeRecords(bufferEntity);
        } 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