dts数据同步切换模式

kxs-transfer/src/main/java/com/kxs/transfer/api/config/ConsumerAssignConfig.java

@@ -71,7 +71,7 @@ public class ConsumerAssignConfig implements ApplicationListener<ContextRefreshe
                 dtsProperties.getUsername(),
                 dtsProperties.getPassword(),
                 dtsProperties.getInitCheckpoint(),
-                ConsumerContext.ConsumerSubscribeMode.ASSIGN);
+                ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE);
 
         //if this parameter is set, force to use the initCheckpoint to initial
         consumerContext.setForceUseCheckpoint(dtsProperties.getIsForceUseInitCheckpoint());

kxs-transfer/src/main/java/com/kxs/transfer/api/service/user/impl/KxsLeaderServiceImpl.java

@@ -62,11 +62,15 @@ public class KxsLeaderServiceImpl extends ServiceImpl<KxsLeaderMapper, KxsLeader
                     break;
             }
         }
+
         if(operation.equals(OperationType.UPDATE)){
             // 更新数据
             baseMapper.updateById(leader);
         }
         if(operation.equals(OperationType.INSERT)){
+            if(id != leader.getUserId()){
+                return;
+            }
             KxsLeader kxsLeader = baseMapper.selectById(leader.getId());
             if (kxsLeader != null){
                 baseMapper.updateById(leader);