public static Optional<DynamicTableSink> getDynamicTableSink(
            ContextResolvedTable contextResolvedTable,
            LogicalTableModify tableModify,
            CatalogManager catalogManager) {
        final FlinkContext context = ShortcutUtils.unwrapContext(tableModify.getCluster());

        CatalogBaseTable catalogBaseTable = contextResolvedTable.getTable();
        // only consider DynamicTableSink
        if (catalogBaseTable instanceof CatalogTable) {
            ResolvedCatalogTable resolvedTable = contextResolvedTable.getResolvedTable();
            Optional<Catalog> optionalCatalog = contextResolvedTable.getCatalog();
            ObjectIdentifier objectIdentifier = contextResolvedTable.getIdentifier();
            boolean isTemporary = contextResolvedTable.isTemporary();
            // only consider the CatalogTable that doesn't use legacy connector sink option
            if (!contextResolvedTable.isAnonymous()
                    && !TableFactoryUtil.isLegacyConnectorOptions(
                            catalogManager
                                    .getCatalog(objectIdentifier.getCatalogName())
                                    .orElse(null),
                            context.getTableConfig(),
                            !context.isBatchMode(),
                            objectIdentifier,
                            resolvedTable,
                            isTemporary)) {
                // create table dynamic table sink
                DynamicTableSink tableSink =
                        ExecutableOperationUtils.createDynamicTableSink(
                                optionalCatalog.orElse(null),
                                () ->
                                        context.getModuleManager()
                                                .getFactory((Module::getTableSinkFactory)),
                                objectIdentifier,
                                resolvedTable,
                                Collections.emptyMap(),
                                context.getTableConfig(),
                                context.getClassLoader(),
                                contextResolvedTable.isTemporary());
                return Optional.of(tableSink);
            }
        }
        return Optional.empty();
    }
@Test
    public void testGetDynamicTableSink() {
        // create a table with connector = test-update-delete
        Map<String, String> options = new HashMap<>();
        options.put("connector", "test-update-delete");
        CatalogTable catalogTable = createTestCatalogTable(options);
        ObjectIdentifier tableId = ObjectIdentifier.of("builtin", "default", "t");
        catalogManager.createTable(catalogTable, tableId, false);
        ContextResolvedTable resolvedTable =
                ContextResolvedTable.permanent(
                        tableId, catalog, catalogManager.resolveCatalogTable(catalogTable));
        LogicalTableModify tableModify = getTableModifyFromSql("DELETE FROM t");
        Optional<DynamicTableSink> optionalDynamicTableSink =
                DeletePushDownUtils.getDynamicTableSink(resolvedTable, tableModify, catalogManager);
        // verify we can get the dynamic table sink
        assertThat(optionalDynamicTableSink).isPresent();
        assertThat(optionalDynamicTableSink.get())
                .isInstanceOf(TestUpdateDeleteTableFactory.SupportsDeletePushDownSink.class);

        // create table with connector = COLLECTION, it's legacy table sink
        options.put("connector", "COLLECTION");
        catalogTable = createTestCatalogTable(options);
        tableId = ObjectIdentifier.of("builtin", "default", "t1");
        catalogManager.createTable(catalogTable, tableId, false);
        resolvedTable =
                ContextResolvedTable.permanent(
                        tableId, catalog, catalogManager.resolveCatalogTable(catalogTable));
        tableModify = getTableModifyFromSql("DELETE FROM t1");
        optionalDynamicTableSink =
                DeletePushDownUtils.getDynamicTableSink(resolvedTable, tableModify, catalogManager);
        // verify it should be empty since it's not an instance of DynamicTableSink but is legacy
        // TableSink
        assertThat(optionalDynamicTableSink).isEmpty();
    }