优化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.SUBSCRIBE);
+                ConsumerContext.ConsumerSubscribeMode.ASSIGN);
 
         //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/config/UserMetaStore.java

@@ -9,6 +9,10 @@ import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.kxs.transfer.api.model.KxsCheckpoint;
 import com.kxs.transfer.api.service.KxsCheckpointService;
 
+import java.util.Comparator;
+import java.util.List;
+import java.util.Optional;
+
 /**
  * store the checkpoint data in the shared storage, such us database, shared file storage...
  * this meta store need to be completed by consumer
@@ -44,12 +48,12 @@ public class UserMetaStore extends AbstractUserMetaStore {
     @Override
     protected String getData(String groupId) {
 
-        KxsCheckpoint checkpoint = kxsCheckpointService.getOne(Wrappers.lambdaQuery(KxsCheckpoint.class)
+        List<KxsCheckpoint> checkpoints = kxsCheckpointService.list(Wrappers.lambdaQuery(KxsCheckpoint.class)
                 .eq(KxsCheckpoint::getGroupId, groupId)
                 .orderByDesc(KxsCheckpoint::getCreateTime)
-                .last("limit 1"));
-
+                .last("limit 5"));
+        Optional<KxsCheckpoint> checkpoint = checkpoints.stream().min(Comparator.comparing(KxsCheckpoint::getCreateTime));
 
-        return checkpoint != null ? checkpoint.getCheckpoint() : null;
+        return checkpoint.map(KxsCheckpoint::getCheckpoint).orElse(null);
     }
 }