@SuppressWarnings({"rawtypes"})
    private void flushSingleTable(String tableIdentifier, List values) {
        if (checkFlushException(tableIdentifier) || values == null || values.isEmpty()) {
            return;
        }
        String loadValue = null;
        RespContent respContent;
        try {
            // support csv and json format
            String format = executionOptions.getStreamLoadProp().getProperty(FORMAT_KEY, FORMAT_JSON_VALUE);
            loadValue = serialize(tableIdentifier, values, format);
            respContent = load(tableIdentifier, loadValue);
            try {
                if (null != metricData && null != respContent) {
                    if (multipleSink) {
                        String[] tableWithDb = tableIdentifier.split("\\.");
                        metricData.outputMetrics(tableWithDb[0], tableWithDb[1],
                                respContent.getNumberLoadedRows(), respContent.getLoadBytes());
                    } else {
                        metricData.invoke(respContent.getNumberLoadedRows(), respContent.getLoadBytes());
                    }
                }
            } catch (Exception e) {
                LOG.warn("metricData invoke get err:", e);
            }
            writeOutNum.addAndGet(values.size());
            // Clean the data that has been loaded.
            values.clear();
        } catch (Exception e) {
            LOG.error(String.format("Flush table: %s error", tableIdentifier), e);
            flushExceptionMap.put(tableIdentifier, e);
            // may count repeatedly
            errorNum.getAndAdd(values.size());

            if (!multipleSink) {
                try {
                    handleSingleTable(e, values, loadValue);
                    return;
                } catch (Exception ex) {
                    throw new RuntimeException(e);
                }
            }

            if (SchemaUpdateExceptionPolicy.THROW_WITH_STOP == schemaUpdatePolicy) {
                throw new RuntimeException(
                        String.format("Writing records to streamload of tableIdentifier:%s failed, the value: %s.",
                                tableIdentifier, loadValue),
                        e);
            }
            if (SchemaUpdateExceptionPolicy.STOP_PARTIAL == schemaUpdatePolicy) {
                errorTables.add(tableIdentifier);
                LOG.warn("The tableIdentifier: {} load failed and the data will be throw away in the future "
                        + "because the option 'sink.multiple.schema-update.policy' is 'STOP_PARTIAL'",
                        tableIdentifier);
                return;
            }
            if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                errorTables.add(tableIdentifier);
                // archive dirty data when 'sink.multiple.schema-update.policy' is 'LOG_WITH_IGNORE'
                for (Object value : values) {
                    try {
                        handleDirtyData(OBJECT_MAPPER.readTree(OBJECT_MAPPER.writeValueAsString(value)),
                                DirtyType.BATCH_LOAD_ERROR, e);
                    } catch (Exception ex) {
                        if (!dirtySinkHelper.getDirtyOptions().ignoreSideOutputErrors()) {
                            throw new RuntimeException(ex);
                        }
                        LOG.warn("Dirty sink failed", ex);
                    }
                }
            }

            values.clear();
        }
    }
A builder used to set parameters to the output format's configuration in a fluent way.

@return builder
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java