diff --git a/pom.xml b/pom.xml
index 6f5c79a8ebd61..e2de9d182d914 100644
--- a/pom.xml
+++ b/pom.xml
@@ -89,7 +89,7 @@
         
         <junit.version>4.12</junit.version>
         <hamcrest.version>1.3</hamcrest.version>
-        <mockito.version>3.4.0</mockito.version>
+        <mockito.version>3.4.2</mockito.version>
         <logback.version>1.2.0</logback.version>
         
         <hikari-cp.version>3.4.2</hikari-cp.version>
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/pom.xml b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/pom.xml
index 75578e78ee0be..d6179365f0aed 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/pom.xml
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/pom.xml
@@ -50,5 +50,11 @@
             <artifactId>h2</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.shardingsphere</groupId>
+            <artifactId>shardingsphere-jdbc-core</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparerTest.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparerTest.java
index dde753749353c..bfbc190245711 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparerTest.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-dialect/shardingsphere-data-pipeline-mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/prepare/datasource/MySQLDataSourcePreparerTest.java
@@ -19,22 +19,28 @@
 
 import org.apache.shardingsphere.data.pipeline.api.config.rulealtered.PipelineConfiguration;
 import org.apache.shardingsphere.data.pipeline.api.datanode.JobDataNodeLine;
+import org.apache.shardingsphere.data.pipeline.api.datasource.PipelineDataSourceWrapper;
+import org.apache.shardingsphere.data.pipeline.api.datasource.config.PipelineDataSourceConfigurationFactory;
 import org.apache.shardingsphere.data.pipeline.api.datasource.config.impl.ShardingSpherePipelineDataSourceConfiguration;
 import org.apache.shardingsphere.data.pipeline.api.datasource.config.yaml.YamlPipelineDataSourceConfiguration;
-import org.apache.shardingsphere.data.pipeline.core.datasource.creator.PipelineDataSourceCreatorFactory;
+import org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
 import org.apache.shardingsphere.data.pipeline.core.exception.PipelineJobPrepareFailedException;
 import org.apache.shardingsphere.data.pipeline.core.prepare.datasource.PrepareTargetTablesParameter;
+import org.apache.shardingsphere.driver.config.datasource.ShardingSpherePipelineDataSourceCreator;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
+import org.mockito.MockedStatic;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import javax.sql.DataSource;
 import java.sql.SQLException;
 import java.util.Collections;
 
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -60,38 +66,56 @@ public final class MySQLDataSourcePreparerTest {
     @Mock
     private ShardingSpherePipelineDataSourceConfiguration targetScalingDataSourceConfig;
     
-    @Mock(extraInterfaces = AutoCloseable.class)
-    private DataSource sourceDataSource;
+    @Mock
+    private PipelineDataSourceWrapper sourceDataSourceWrapper;
     
-    @Mock(extraInterfaces = AutoCloseable.class)
-    private DataSource targetDataSource;
+    @Mock
+    private PipelineDataSourceWrapper targetDataSourceWrapper;
+    
+    @Mock
+    private ShardingSpherePipelineDataSourceCreator mockShardingSpherePipelineDataSourceCreator;
     
     @Before
     public void setUp() throws SQLException {
-        when(prepareTargetTablesParameter.getPipelineConfiguration()).thenReturn(pipelineConfig);
-        when(prepareTargetTablesParameter.getTablesFirstDataNodes()).thenReturn(new JobDataNodeLine(Collections.emptyList()));
+        PipelineDataSourceManager mockPipelineDataSourceManager = mock(PipelineDataSourceManager.class);
+        when(mockPipelineDataSourceManager.getDataSource(same(sourceScalingDataSourceConfig))).thenReturn(sourceDataSourceWrapper);
+        when(mockPipelineDataSourceManager.getDataSource(same(targetScalingDataSourceConfig))).thenReturn(targetDataSourceWrapper);
+        when(prepareTargetTablesParameter.getDataSourceManager()).thenReturn(mockPipelineDataSourceManager);
         when(pipelineConfig.getSource()).thenReturn(sourceYamlPipelineDataSourceConfiguration);
-        when(PipelineDataSourceCreatorFactory.getInstance(
-                sourceScalingDataSourceConfig.getType()).createPipelineDataSource(sourceScalingDataSourceConfig.getDataSourceConfiguration())).thenReturn(sourceDataSource);
+        when(pipelineConfig.getSource().getType()).thenReturn("ShardingSphereJDBC");
+        when(pipelineConfig.getSource().getParameter()).thenReturn("source");
         when(pipelineConfig.getTarget()).thenReturn(targetYamlPipelineDataSourceConfiguration);
-        when(PipelineDataSourceCreatorFactory.getInstance(
-                targetScalingDataSourceConfig.getType()).createPipelineDataSource(targetScalingDataSourceConfig.getDataSourceConfiguration())).thenReturn(targetDataSource);
-    }
+        when(pipelineConfig.getTarget().getType()).thenReturn("ShardingSphereJDBC");
+        when(pipelineConfig.getTarget().getParameter()).thenReturn("target");
+        when(prepareTargetTablesParameter.getPipelineConfiguration()).thenReturn(pipelineConfig);
+        when(prepareTargetTablesParameter.getTablesFirstDataNodes()).thenReturn(new JobDataNodeLine(Collections.emptyList()));
+    } 
     
     @Test
-    @Ignore
     public void assertGetConnection() throws SQLException {
-        MySQLDataSourcePreparer mySQLDataSourcePreparer = new MySQLDataSourcePreparer();
-        mySQLDataSourcePreparer.prepareTargetTables(prepareTargetTablesParameter);
-        verify(sourceDataSource).getConnection();
-        verify(targetDataSource).getConnection();
+        try (MockedStatic<PipelineDataSourceConfigurationFactory> mockedStaticPipelineDataSourceConfigurationFactory = mockStatic(PipelineDataSourceConfigurationFactory.class);) {
+            mockedStaticPipelineDataSourceConfigurationFactory.when(() -> PipelineDataSourceConfigurationFactory.newInstance(eq("ShardingSphereJDBC"), eq("source")))
+                    .thenReturn(sourceScalingDataSourceConfig);
+            mockedStaticPipelineDataSourceConfigurationFactory.when(() -> PipelineDataSourceConfigurationFactory.newInstance(eq("ShardingSphereJDBC"), eq("target")))
+                    .thenReturn(targetScalingDataSourceConfig);
+            MySQLDataSourcePreparer mySQLDataSourcePreparer = new MySQLDataSourcePreparer();
+            mySQLDataSourcePreparer.prepareTargetTables(prepareTargetTablesParameter);
+            verify(sourceDataSourceWrapper).getConnection();
+            verify(targetDataSourceWrapper).getConnection();
+        }
     }
     
     @Test(expected = PipelineJobPrepareFailedException.class)
-    @Ignore
     public void assertThrowPrepareFailedException() throws SQLException {
-        when(sourceDataSource.getConnection()).thenThrow(SQLException.class);
-        MySQLDataSourcePreparer mySQLDataSourcePreparer = new MySQLDataSourcePreparer();
-        mySQLDataSourcePreparer.prepareTargetTables(prepareTargetTablesParameter);
+        try (MockedStatic<PipelineDataSourceConfigurationFactory> mockedStaticPipelineDataSourceConfigurationFactory = mockStatic(PipelineDataSourceConfigurationFactory.class)) {
+            mockedStaticPipelineDataSourceConfigurationFactory.when(() -> PipelineDataSourceConfigurationFactory.newInstance(eq("ShardingSphereJDBC"), eq("source")))
+                    .thenReturn(sourceScalingDataSourceConfig);
+            mockedStaticPipelineDataSourceConfigurationFactory.when(() -> PipelineDataSourceConfigurationFactory.newInstance(eq("ShardingSphereJDBC"), eq("target")))
+                    .thenReturn(targetScalingDataSourceConfig);
+            when(sourceDataSourceWrapper.getConnection()).thenThrow(SQLException.class);
+            MySQLDataSourcePreparer mySQLDataSourcePreparer = new MySQLDataSourcePreparer();
+            mySQLDataSourcePreparer.prepareTargetTables(prepareTargetTablesParameter);
+        }
+       
     }
 }