@Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        final TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);

        final ReadableConfig tableOptions = helper.getOptions();

        final EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat = getValueEncodingFormat(helper);

        // validate all options
        helper.validateExcept(ExtractNode.INLONG_MSG);

        validatePKConstraints(context.getObjectIdentifier(), context.getCatalogTable(), valueEncodingFormat);

        final Configuration properties = getTubeMQProperties(context.getCatalogTable().getOptions());

        final DataType physicalDataType = context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType();

        return createTubeMQTableSink(
                physicalDataType,
                valueEncodingFormat,
                TubeMQOptions.getSinkTopics(tableOptions),
                TubeMQOptions.getMasterRpcAddress(tableOptions),
                TubeMQOptions.getTiSet(tableOptions),
                properties);
    }
A dynamic table factory implementation for TubeMQ.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/tubemq/src/main/java/org/apache/inlong/sort/tubemq/table/TubeMQDynamicTableFactory.java