private DataSource createDataSource(
            SourceDef sourceDef, StreamExecutionEnvironment env, Configuration pipelineConfig) {
        // Search the data source factory
        DataSourceFactory sourceFactory =
                FactoryDiscoveryUtils.getFactoryByIdentifier(
                        sourceDef.getType(), DataSourceFactory.class);
        // Add source JAR to environment
        FactoryDiscoveryUtils.getJarPathByIdentifier(sourceFactory)
                .ifPresent(jar -> FlinkEnvironmentUtils.addJar(env, jar));
        DataSource dataSource =
                sourceFactory.createDataSource(
                        new FactoryHelper.DefaultContext(
                                sourceDef.getConfig(),
                                pipelineConfig,
                                Thread.currentThread().getContextClassLoader()));
        return dataSource;
    }
@Test
    void testCreateDataSourceFromSourceDef() {
        SourceDef sourceDef =
                new SourceDef(
                        "data-source-factory-1",
                        "source-database",
                        Configuration.fromMap(
                                ImmutableMap.<String, String>builder()
                                        .put("host", "0.0.0.0")
                                        .build()));

        DataSourceFactory sourceFactory =
                FactoryDiscoveryUtils.getFactoryByIdentifier(
                        sourceDef.getType(), DataSourceFactory.class);

        DataSource dataSource =
                sourceFactory.createDataSource(
                        new FactoryHelper.DefaultContext(
                                sourceDef.getConfig(),
                                new Configuration(),
                                Thread.currentThread().getContextClassLoader()));

        Assertions.assertTrue(dataSource instanceof DataSourceFactory1.TestDataSource);
        Assertions.assertEquals(
                "0.0.0.0", ((DataSourceFactory1.TestDataSource) dataSource).getHost());
    }