Merge branch 'dmp_dev' of http://gitlab.ioubuy.cn/yaobenzhang/jz-dmp-service.git into dmp_dev

src/main/java/com/jz/dmp/modules/controller/DataIntegration/RealTimeSyncController.java

@@ -129,9 +129,10 @@ public class RealTimeSyncController {
      */
     @ApiOperation(value = "获取数据源和目标数据源下拉框", notes = "获取数据源和目标数据源下拉框")
     @GetMapping(value = "/getDatasourceNameList")
-    @ApiImplicitParams({@ApiImplicitParam(name = "projectId", value = "项目id",required = true),
-            @ApiImplicitParam(name = "type", value = "数据源类型：01来源，02 目标源",required = true)})
+    @ApiImplicitParams({@ApiImplicitParam(name = "projectId", value = "项目id", required = true),
+            @ApiImplicitParam(name = "type", value = "数据源类型：01来源，02 目标源", required = true)})
     public JsonResult<List<DataSourceNameListDto>> getDatasourceNameList(@RequestParam String projectId, @RequestParam String type) throws Exception {
+        logger.info("###################请求参数{}projectId=" + projectId + "&type=" + type);
         if (StringUtils.isEmpty(projectId)) {
             return new JsonResult(ResultCode.PARAMS_ERROR, "项目id不能为空！");
         }
@@ -149,9 +150,10 @@ public class RealTimeSyncController {
      * @author Bellamy
      * @since 2021-01-06
      */
-    @ApiOperation(value = "新增--选择源表信息", notes = "新增--选择源表信息")
+    @ApiOperation(value = "新增--选择源表信息和配置目标表", notes = "新增--选择源表信息和配置目标表")
     @PostMapping(value = "/getSourceDbTableList")
     public JsonResult<RealTimeSyncModel> getTableInfo(@RequestBody @Validated RealTimeTableInfoReq req, HttpServletRequest httpRequest) throws Exception {
+        logger.info("###################请求参数{}" + req.toString() + "############");
         if (StringUtils.isEmpty(req.getProjectId())) {
             return new JsonResult(ResultCode.PARAMS_ERROR, "项目id不能为空！");
         }
@@ -171,7 +173,7 @@ public class RealTimeSyncController {
      */
     @ApiOperation(value = "新增--选择目标信息", notes = "新增--选择目标信息")
     @GetMapping(value = "/targetDatasourceInfo")
-    @ApiImplicitParam(name = "projectId", value = "项目id",required = true)
+    @ApiImplicitParam(name = "projectId", value = "项目id", required = true)
     public JsonResult getTargetDatasourceInfo(@RequestParam String projectId) throws Exception {
         if (StringUtils.isEmpty(projectId)) {
             return new JsonResult(ResultCode.PARAMS_ERROR, "项目id不能为空！");
@@ -180,4 +182,27 @@ public class RealTimeSyncController {
         return jsonResult;
     }
 
+    /**
+     * 设置成黑名单时 ,检查是否有同步任务存在运行
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-01-07
+     */
+    @ApiOperation(value = "设置成黑名单时 ,检查是否有同步任务存在运行", notes = "设置成黑名单时 ,检查是否有同步任务存在运行")
+    @GetMapping(value = "/settingBlackListCheckTask")
+    @ApiImplicitParams({@ApiImplicitParam(name = "srcDatasourceId", value = "来源数据源id", required = true),
+            @ApiImplicitParam(name = "sourceTableName", value = "来源表名称", required = true)})
+    public JsonResult settingBlackListCheckTask(@RequestParam String srcDatasourceId, @RequestParam String sourceTableName) throws Exception {
+        logger.info("###################请求参数{}srcDatasourceId=" + srcDatasourceId + "&sourceTableName=" + sourceTableName);
+        if (StringUtils.isEmpty(srcDatasourceId)) {
+            return new JsonResult(ResultCode.PARAMS_ERROR, "来源数据源id不能为空！");
+        }
+        if (StringUtils.isEmpty(sourceTableName)) {
+            return new JsonResult(ResultCode.PARAMS_ERROR, "来源表名称不能为空！");
+        }
+        JsonResult jsonResult = dmpRealtimeSyncInfoService.queryRealTimeInfoByDataSourceId(srcDatasourceId, sourceTableName);
+        return jsonResult;
+    }
+
 }
\ No newline at end of file

src/main/java/com/jz/dmp/modules/dao/DmpRealtimeSyncInfoDao.java

@@ -112,7 +112,14 @@ public interface DmpRealtimeSyncInfoDao {
      */
     RealTimeSyncDataSourceModel querygSourceDbInfoById(@Param("srcDataSourceId") String srcDataSourceId) throws Exception;
 
-    List<Map> queryRealTimeInfoByDataSourceId(@Param("srcDatasourceId") String srcDatasourceId, @Param("targetDatasourceId") String targetDatasourceId);
+    /**
+     * 数据源对应的表详细信息
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-01-06
+     */
+    List<Map> queryRealTimeInfoByDataSourceId(Map map) throws Exception;
 
     /**
      * 查询源数据源的黑名单表

src/main/java/com/jz/dmp/modules/service/DmpRealtimeSyncInfoService.java

@@ -100,4 +100,6 @@ public interface DmpRealtimeSyncInfoService {
      * @since 2021-01-07
      */
     JsonResult selectTargetDatasourceInfo(String projectId) throws Exception;
+
+    JsonResult queryRealTimeInfoByDataSourceId(String srcDatasourceId, String sourceTableName) throws Exception;
 }
\ No newline at end of file

src/main/java/com/jz/dmp/modules/service/impl/DmpRealtimeSyncInfoServiceImpl.java

@@ -180,8 +180,12 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
         //根据来源数据源id获取数据信息
         RealTimeSyncDataSourceModel sourceDbInfo = dmpRealtimeSyncInfoDao.querygSourceDbInfoById(req.getSrcDatasourceId());
         sourceDbInfo.setPassword(new BaseService().decode(sourceDbInfo.getPassword(), publicKey));
+
+        Map map = new HashMap();
+        map.put("srcDatasourceId", req.getSrcDatasourceId());
+        map.put("targetDatasourceId", req.getTargetDatasourceId());
         //数据源对应的表详细信息  判断表的记录在同步任务中是否存在记录,如果已经存在,说明同步过
-        List<Map> list = dmpRealtimeSyncInfoDao.queryRealTimeInfoByDataSourceId(req.getSrcDatasourceId(), req.getTargetDatasourceId());
+        List<Map> list = dmpRealtimeSyncInfoDao.queryRealTimeInfoByDataSourceId(map);
         if (list.size() > 0 && list != null) {
             paramsMap = new HashMap<>();
             for (Map item : list) {
@@ -226,7 +230,27 @@ public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoServic
     public JsonResult selectTargetDatasourceInfo(String projectId) throws Exception {
         DmpProjectSystemInfo dmpProjectSystemInfo = dmpProjectDao.queryProjectSystemInfo(Long.valueOf(projectId));
         String kafkaConnectUrl = dmpProjectSystemInfo.getKafkaConnectorUrl(); //kafka 连接信息
-        String [] arr = kafkaConnectUrl.split(",");
+        String[] arr = kafkaConnectUrl.split(",");
         return new JsonResult(arr);
     }
+
+    /**
+     * 设置成黑名单时 ,检查是否有同步任务存在运行
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-01-07
+     */
+    @Override
+    public JsonResult queryRealTimeInfoByDataSourceId(String srcDatasourceId, String sourceTableName) throws Exception {
+        Map map = new HashMap();
+        map.put("srcDatasourceId", srcDatasourceId); //来源表 数据源id
+        map.put("sourceTableName", sourceTableName); // 来源表
+        List<Map> list = dmpRealtimeSyncInfoDao.queryRealTimeInfoByDataSourceId(map);
+        boolean flag = false;
+        if (list.size() > 0 && list != null) {
+            flag = true;
+        }
+        return new JsonResult(flag);
+    }
 }
\ No newline at end of file

src/main/resources/mapper/dmp/DmpRealtimeSyncInfoMapper.xml

@@ -380,7 +380,7 @@
         and  ds.ID = #{srcDataSourceId}
     </select>
 
-    <select id="queryRealTimeInfoByDataSourceId" resultType="java.util.Map">
+    <select id="queryRealTimeInfoByDataSourceId" resultType="java.util.Map" parameterType="java.util.Map">
         select
             src_table_name as srcTableName,
             connector_job_id as connectorJobId,
@@ -390,7 +390,7 @@
         where 1=1 and type =2
         and src_datasource_id = #{srcDatasourceId}
         <if test="targetDatasourceId != null"> and target_datasource_id = #{targetDatasourceId} </if>
-        <!--<if test="srcTableName != null"> and src_table_name = #{srcTableName} </if>-->
+        <if test="sourceTableName != null and sourceTableName !=''"> and src_table_name = #{sourceTableName} </if>
     </select>
 
     <!-- 查询源数据源的黑名单表 -->