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

        final ReadableConfig tableOptions = helper.getOptions();

        final DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat = getValueDecodingFormat(helper);

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

        validatePKConstraints(context.getObjectIdentifier(), context.getCatalogTable(), valueDecodingFormat);
        innerFormat = tableOptions.get(FORMAT).contains(ExtractNode.INLONG_MSG);

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

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

        String inlongMetric = tableOptions.getOptional(INLONG_METRIC).orElse(null);
        String auditHostAndPorts = tableOptions.get(INLONG_AUDIT);
        String auditKeys = tableOptions.get(AUDIT_KEYS);

        final DirtyOptions dirtyOptions = DirtyOptions.fromConfig(tableOptions);
        final DirtySink<byte[]> dirtySink = DirtySinkFactoryUtils.createDirtySink(context, dirtyOptions);

        return createTubeMQTableSource(
                physicalDataType,
                valueDecodingFormat,
                TubeMQOptions.getSourceTopics(tableOptions),
                TubeMQOptions.getMasterRpcAddress(tableOptions),
                TubeMQOptions.getTiSet(tableOptions),
                TubeMQOptions.getConsumerGroup(tableOptions),
                TubeMQOptions.getSessionKey(tableOptions),
                properties,
                inlongMetric,
                auditHostAndPorts,
                auditKeys,
                dirtySink,
                dirtyOptions);
    }
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