public final synchronized void writeRecords(String database,
            String table,
            List<Map<String, String>> records,
            String dirtyLogTag,
            String dirtyIdentify,
            String dirtyLabel)
            throws IOException {
        checkFlushException();
        try {
            if (0 == records.size()) {
                return;
            }
            String bufferKey = String.format("%s,%s", database, table);
            SinkBufferEntity bufferEntity = bufferMap.computeIfAbsent(bufferKey,
                    k -> new SinkBufferEntity(database, table, sinkOptions.getLabelPrefix()));
            Map<String, String> head = records.get(0);
            String columns = StringUtils.join(head.keySet(), ",");
            bufferEntity.setColumns(columns);
            bufferEntity.setDirtyLogTag(dirtyLogTag);
            bufferEntity.setDirtyIdentify(dirtyIdentify);
            bufferEntity.setDirtyLabel(dirtyLabel);

            String columnSeparator = StarRocksDelimiterParser.parse(
                    sinkOptions.getSinkStreamLoadProperties().get("column_separator"), "\t");
            if (StarRocksSinkOptions.StreamLoadFormat.CSV.equals(sinkOptions.getStreamLoadFormat())) {
                for (Map<String, String> record : records) {
                    StringBuilder sb = new StringBuilder();
                    int idx = 0;
                    int len = record.values().size();
                    for (String val : record.values()) {
                        sb.append(null == val ? "\\N" : val);
                        if (idx++ < len - 1) {
                            sb.append(columnSeparator);
                        }
                    }
                    byte[] bts = sb.toString().getBytes(StandardCharsets.UTF_8);
                    bufferEntity.addToBuffer(bts);
                }
            } else {
                for (Map<String, String> record : records) {
                    byte[] bts = JSON.toJSONString(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