public static Table getStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExecutionEnvironment env,
                                        StreamTableEnvironment tableEnv, String sqlRootDir, String pluginLoadMode) throws Exception {

        String sourceTypeStr = sourceTableInfo.getType();
        String typeNoVersion = DtStringUtil.getPluginTypeWithoutVersion(sourceTypeStr);
        String pluginJarPath = PluginUtil.getJarFileDirPath(String.format(DIR_NAME_FORMAT, sourceTypeStr), sqlRootDir, pluginLoadMode);
        String className = PluginUtil.getGenerClassName(typeNoVersion, CURR_TYPE);

        return ClassLoaderManager.newInstance(pluginJarPath, (cl) -> {
            Class<?> sourceClass = cl.loadClass(className);
            if(!IStreamSourceGener.class.isAssignableFrom(sourceClass)){
                throw new RuntimeException("class " + sourceClass.getName() + " not subClass of IStreamSourceGener");
            }

            IStreamSourceGener sourceGener = sourceClass.asSubclass(IStreamSourceGener.class).newInstance();
            return (Table) sourceGener.genStreamSource(sourceTableInfo, env, tableEnv);
        });
    }
The configuration of the type specified data source
@param sourceTableInfo
@return
core/src/main/java/com/dtstack/flink/sql/source/StreamSourceFactory.java