public static <T> DirtySink<T> createDirtySink(Context context, DirtyOptions dirtyOptions) {
        if (dirtyOptions == null) {
            dirtyOptions = DirtyOptions.fromConfig(Configuration.fromMap(context.getCatalogTable().getOptions()));
        }
        dirtyOptions.validate();
        DirtySink<T> dirtySink = null;
        if (dirtyOptions.ignoreDirty() && dirtyOptions.enableDirtySideOutput()) {
            DirtySinkFactory dirtySinkFactory = FactoryUtil.discoverFactory(context.getClassLoader(),
                    DirtySinkFactory.class, dirtyOptions.getDirtyConnector());
            dirtySink = dirtySinkFactory.createDirtySink(context);
        }
        return dirtySink;
    }
Dirty sink facotry utils, it helps to create dirty sink
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/dirty/utils/DirtySinkFactoryUtils.java