@Override
    public CDCJobConfiguration swapToObject(final YamlCDCJobConfiguration yamlConfig) {
        List<JobDataNodeLine> jobShardingDataNodes = null == yamlConfig.getJobShardingDataNodes()
                ? Collections.emptyList()
                : yamlConfig.getJobShardingDataNodes().stream().map(JobDataNodeLine::unmarshal).collect(Collectors.toList());
        YamlSinkConfiguration yamlSinkConfig = yamlConfig.getSinkConfig();
        SinkConfiguration sinkConfig = new SinkConfiguration(CDCSinkType.valueOf(yamlSinkConfig.getSinkType()), yamlSinkConfig.getProps());
        JobDataNodeLine tablesFirstDataNodes = null == yamlConfig.getTablesFirstDataNodes() ? null : JobDataNodeLine.unmarshal(yamlConfig.getTablesFirstDataNodes());
        return new CDCJobConfiguration(yamlConfig.getJobId(), yamlConfig.getDatabaseName(), yamlConfig.getSchemaTableNames(), yamlConfig.isFull(),
                TypedSPILoader.getService(DatabaseType.class, yamlConfig.getSourceDatabaseType()),
                (ShardingSpherePipelineDataSourceConfiguration) dataSourceConfigSwapper.swapToObject(yamlConfig.getDataSourceConfiguration()), tablesFirstDataNodes,
                jobShardingDataNodes, yamlConfig.isDecodeWithTX(), sinkConfig, yamlConfig.getConcurrency(), yamlConfig.getRetryTimes());
    }
@Test
    void assertSwapToObject() {
        YamlCDCJobConfiguration yamlJobConfig = new YamlCDCJobConfiguration();
        yamlJobConfig.setJobId("j0302p00007a8bf46da145dc155ba25c710b550220");
        yamlJobConfig.setDatabaseName("test_db");
        yamlJobConfig.setSchemaTableNames(Arrays.asList("test.t_order", "t_order_item"));
        yamlJobConfig.setFull(true);
        yamlJobConfig.setSourceDatabaseType("MySQL");
        YamlSinkConfiguration sinkConfig = new YamlSinkConfiguration();
        sinkConfig.setSinkType(CDCSinkType.SOCKET.name());
        yamlJobConfig.setSinkConfig(sinkConfig);
        CDCJobConfiguration actual = new YamlCDCJobConfigurationSwapper().swapToObject(yamlJobConfig);
        assertThat(actual.getJobId(), is("j0302p00007a8bf46da145dc155ba25c710b550220"));
        assertThat(actual.getDatabaseName(), is("test_db"));
        assertThat(actual.getSchemaTableNames(), is(Arrays.asList("test.t_order", "t_order_item")));
        assertTrue(actual.isFull());
    }