init

fn-opensearch/src/main/java/net/yaoyi/pipeline/opensearch/OpenSearchHandler.java

@@ -39,14 +39,23 @@ public class OpenSearchHandler implements PojoRequestHandler<TaskImageMsg, TaskI
             return new TaskImageResultMsg();
         }
 
-        try {
-            Thread.sleep(2000);
-        } catch (Exception e) {
-            log.warn("等待被中断：" + e.getMessage());
+        List<TaskImageIndex> list = Collections.emptyList();
+        int times = 5;
+        while (times > 0) {
+            try {
+                Thread.sleep(1000);
+            } catch (Exception e) {
+                log.warn("等待被中断：" + e.getMessage());
+            }
+            list = fetchByTaskId(taskImageMsg.getIndexTableName(), namespace, taskId);
+            if (dataQuerySuccess(taskImageMsg.getImages(), list)) {
+                break;
+            } else {
+                times--;
+            }
         }
 
         // 2. 以 namespace + taskId 维度来完成数据更新，image相同的数据保持id不变
-        List<TaskImageIndex> list = fetchByTaskId(taskImageMsg.getIndexTableName(), namespace, taskId);
         for (TaskImageIndex taskImageIndex : list) {
             if (taskImageIndex.getVector() == null) {
                 throw new RuntimeException("尚未更新索引信息，id:" + taskImageIndex.getId());
@@ -81,6 +90,16 @@ public class OpenSearchHandler implements PojoRequestHandler<TaskImageMsg, TaskI
         return taskImageResultMsg;
     }
 
+    private boolean dataQuerySuccess(List<TaskImageMsg.TaskImage> images, List<TaskImageIndex> list) {
+        if (images.size() != list.size()) {
+            return false;
+        }
+
+        Set<String> set = images.stream().map(TaskImageMsg.TaskImage::getImage).collect(Collectors.toSet());
+        Set<String> set2 = list.stream().map(TaskImageIndex::getImage).collect(Collectors.toSet());
+        return set.removeAll(set2) && set.isEmpty();
+    }
+
     private void saveByTask(TaskImageMsg taskImageMsg) {
         List<TaskImageIndex> addList = new ArrayList<>();
         List<TaskImageIndex> deleteList = new ArrayList<>();