@Override
    public synchronized void invoke(DirtyData<T> dirtyData) throws Exception {
        try {
            addBatch(dirtyData);
        } catch (Exception e) {
            if (!s3Options.ignoreSideOutputErrors()) {
                throw new RuntimeException(String.format("Add batch to identifier:%s failed, the dirty data: %s.",
                        dirtyData.getIdentifier(), dirtyData.toString()), e);
            }
            LOGGER.warn("Add batch to identifier:{} failed "
                    + "and the dirty data will be throw away in the future"
                    + " because the option 'dirty.side-output.ignore-errors' is 'true'", dirtyData.getIdentifier());
        }
        if (buffered() && valid() && !flushing) {
            flush();
        }
    }
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