private void handleDirtyData(String databaseName,
            String tableName,
            int recordNum,
            int recordSize,
            JsonNode data,
            JsonDynamicSchemaFormat jsonFormat,
            Exception e) {
        // upload metrics for dirty data
        if (null != metricData) {
            if (sinkMultipleEnable) {
                metricData.outputDirtyMetrics(databaseName, tableName, recordNum, recordSize);
            } else {
                metricData.invokeDirty(recordNum, recordSize);
            }
        }

        if (!dirtyOptions.ignoreDirty()) {
            return;
        }
        if (data == null || jsonFormat == null) {
            return;
        }
        Triple<String, String, String> triple = getDirtyLabelTagAndIdentity(data, jsonFormat);
        String label = triple.getLeft();
        String tag = triple.getMiddle();
        String identify = triple.getRight();
        if (label == null || tag == null || identify == null) {
            LOG.warn("dirty label or tag or identify is null, ignore dirty data writing");
            return;
        }
        // archive dirty data
        DirtySinkHelper<Object> dirtySinkHelper = new DirtySinkHelper<>(dirtyOptions, dirtySink);
        List<Map<String, Object>> physicalDataList = HiveTableUtil.jsonNode2Map(jsonFormat.getPhysicalData(data));
        for (Map<String, Object> record : physicalDataList) {
            JsonNode jsonNode = HiveTableUtil.object2JsonNode(record);
            dirtySinkHelper.invoke(jsonNode, DirtyType.BATCH_LOAD_ERROR, label, tag, identify, e);
        }
    }
upload dirty data metrics and write dirty data

@param databaseName database name
@param tableName table name
@param recordNum record num
@param recordSize record byte size
@param data raw data
@param jsonFormat json formatter for formatting raw data
@param e exception
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/filesystem/HadoopPathBasedPartFileWriter.java