private void addBatch(DirtyData<T> dirtyData) throws IOException {
        readInNum.incrementAndGet();
        String value;
        Map<String, String> labelMap = LabelUtils.parseLabels(dirtyData.getLabels());
        T data = dirtyData.getData();
        if (data instanceof RowData) {
            value = format((RowData) data, dirtyData.getRowType(), labelMap);
        } else if (data instanceof JsonNode) {
            value = format((JsonNode) data, labelMap);
        } else {
            // Only support csv format when the row is not a 'RowData' and 'JsonNode'
            value = FormatUtils.csvFormat(data, labelMap, s3Options.getFieldDelimiter());
        }
        if (s3Options.enableDirtyLog()) {
            LOGGER.debug("[{}] {}", dirtyData.getLogTag(), value);
        }
        batchBytes += value.getBytes(UTF_8).length;
        size++;
        batchMap.computeIfAbsent(dirtyData.getIdentifier(), k -> new ArrayList<>()).add(value);
    }
S3 dirty sink that is used to sink dirty data to s3

@param <T>
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/sink/s3/S3DirtySink.java