public void invoke(T dirtyData, DirtyType dirtyType, String label, String logTag, String identifier, Throwable e) {
        if (!dirtyOptions.ignoreDirty()) {
            RuntimeException ex;
            if (e instanceof RuntimeException) {
                ex = (RuntimeException) e;
            } else {
                ex = new RuntimeException(e);
            }
            throw ex;
        }
        if (dirtySink != null) {
            DirtyData.Builder<T> builder = DirtyData.builder();
            try {
                builder.setData(dirtyData)
                        .setDirtyType(dirtyType)
                        .setLabels(label)
                        .setLogTag(logTag)
                        .setDirtyMessage(e.getMessage())
                        .setIdentifier(identifier);
                dirtySink.invoke(builder.build());
            } catch (Exception ex) {
                if (!dirtyOptions.ignoreSideOutputErrors()) {
                    throw new RuntimeException(ex);
                }
                LOGGER.warn("Dirty sink failed", ex);
            }
        }
    }
Dirty data sink

@param dirtyData The dirty data
@param dirtyType The dirty type {@link DirtyType}
@param label The dirty label
@param logTag The dirty logTag
@param identifier The dirty identifier
@param e The cause of dirty data
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/DirtySinkHelper.java