[Fix][Connector-v2][Easysearch] Handle optional password in credentials and add clearScroll method (apache#10161)

Author: zhangshenghang

seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/client/EasysearchClient.java

@@ -158,10 +158,14 @@ private static RestClientBuilder getRestClientBuilder(
         restClientBuilder.setHttpClientConfigCallback(
                 httpClientBuilder -> {
                     if (username.isPresent()) {
+                        String passwordStr = null;
+                        if (password.isPresent()) {
+                            passwordStr = password.get();
+                        }
                         CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                         credentialsProvider.setCredentials(
                                 AuthScope.ANY,
-                                new UsernamePasswordCredentials(username.get(), password.get()));
+                                new UsernamePasswordCredentials(username.get(), passwordStr));
                         httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                     }
 
@@ -289,6 +293,36 @@ public void close() {
         }
     }
 
+    public boolean clearScroll(String scrollId) {
+        if (scrollId == null || scrollId.isEmpty()) {
+            return false;
+        }
+
+        String endpoint = "/_search/scroll";
+        Request request = new Request("DELETE", endpoint);
+        Map<String, String> param = new HashMap<>();
+        param.put("scroll_id", scrollId);
+        request.setJsonEntity(JsonUtils.toJsonString(param));
+
+        try {
+            Response response = restClient.performRequest(request);
+            if (response == null) {
+                log.warn("DELETE {} response null when clearing scrollId {}", endpoint, scrollId);
+                return false;
+            }
+            int statusCode = response.getStatusLine().getStatusCode();
+            if (statusCode == HttpStatus.SC_OK) {
+                return true;
+            } else {
+                log.warn("Failed to clear scrollId {}, status code={}", scrollId, statusCode);
+                return false;
+            }
+        } catch (IOException e) {
+            log.warn("Error clearing scrollId " + scrollId, e);
+            return false;
+        }
+    }
+
     /**
      * first time to request search documents by scroll call /${index}/_search?scroll=${scroll}
      *

seatunnel-connectors-v2/connector-easysearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/easysearch/source/EasysearchSourceReader.java

@@ -74,19 +74,33 @@ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
             EasysearchSourceSplit split = splits.poll();
             if (split != null) {
                 SourceIndexInfo sourceIndexInfo = split.getSourceIndexInfo();
-                ScrollResult scrollResult =
-                        ezsClient.searchByScroll(
-                                sourceIndexInfo.getIndex(),
-                                sourceIndexInfo.getSource(),
-                                sourceIndexInfo.getQuery(),
-                                sourceIndexInfo.getScrollTime(),
-                                sourceIndexInfo.getScrollSize());
-                outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
-                while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {
-                    scrollResult =
-                            ezsClient.searchWithScrollId(
-                                    scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());
+                String scrollId = null;
+                try {
+                    ScrollResult scrollResult =
+                            ezsClient.searchByScroll(
+                                    sourceIndexInfo.getIndex(),
+                                    sourceIndexInfo.getSource(),
+                                    sourceIndexInfo.getQuery(),
+                                    sourceIndexInfo.getScrollTime(),
+                                    sourceIndexInfo.getScrollSize());
+                    scrollId = scrollResult.getScrollId();
                     outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
+                    while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {
+                        scrollResult =
+                                ezsClient.searchWithScrollId(
+                                        scrollResult.getScrollId(),
+                                        sourceIndexInfo.getScrollTime());
+                        scrollId = scrollResult.getScrollId();
+                        outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
+                    }
+                } finally {
+                    if (scrollId != null && !scrollId.isEmpty()) {
+                        try {
+                            ezsClient.clearScroll(scrollId);
+                        } catch (Exception e) {
+                            log.warn("Failed to clear Easysearch scrollId: " + scrollId, e);
+                        }
+                    }
                 }
             } else if (noMoreSplit) {
                 // signal to the source that we have reached the end of the data.

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-easysearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/easysearch/EasysearchIT.java

@@ -209,6 +209,7 @@ private List<String> readSinkDataFromIndex(String indexName) throws InterruptedE
         query.put("range", range);
         ScrollResult scrollResult =
                 easysearchClient.searchByScroll(indexName, source, query, "1m", 1000);
+        String scrollId = scrollResult.getScrollId();
         scrollResult
                 .getDocs()
                 .forEach(
@@ -231,6 +232,12 @@ private List<String> readSinkDataFromIndex(String indexName) throws InterruptedE
                                         o -> Integer.valueOf(o.get("c_int").toString())))
                         .map(JsonUtils::toJsonString)
                         .collect(Collectors.toList());
+
+        if (scrollId != null && !scrollId.isEmpty()) {
+            boolean cleared = easysearchClient.clearScroll(scrollId);
+            Assertions.assertTrue(cleared);
+        }
+
         return docs;
     }
 
@@ -344,6 +351,7 @@ private List<String> readSinkData() throws InterruptedException {
         query.put("range", range);
         ScrollResult scrollResult =
                 easysearchClient.searchByScroll("st_index2", source, query, "1m", 1000);
+        String scrollId = scrollResult.getScrollId();
         scrollResult
                 .getDocs()
                 .forEach(
@@ -366,6 +374,12 @@ private List<String> readSinkData() throws InterruptedException {
                                         o -> Integer.valueOf(o.get("c_int").toString())))
                         .map(JsonUtils::toJsonString)
                         .collect(Collectors.toList());
+
+        if (scrollId != null && !scrollId.isEmpty()) {
+            boolean cleared = easysearchClient.clearScroll(scrollId);
+            Assertions.assertTrue(cleared);
+        }
+
         return docs;
     }