[Fix][Connector-V2][Elasticsearch]Fix sink configuration for DROP_DATA (#7124)

Author: Wudadada

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/catalog/ElasticSearchCatalog.java

@@ -217,8 +217,7 @@ public void dropDatabase(TablePath tablePath, boolean ignoreIfNotExists)
 
     @Override
     public void truncateTable(TablePath tablePath, boolean ignoreIfNotExists) {
-        dropTable(tablePath, ignoreIfNotExists);
-        createTable(tablePath, null, ignoreIfNotExists);
+        esRestClient.clearIndexData(tablePath.getTableName());
     }
 
     @Override

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java

@@ -480,6 +480,35 @@ public void dropIndex(String tableName) {
         }
     }
 
+    public void clearIndexData(String indexName) {
+        String endpoint = String.format("/%s/_delete_by_query", indexName);
+        Request request = new Request("POST", endpoint);
+        String jsonString = "{ \"query\": { \"match_all\": {} } }";
+        request.setJsonEntity(jsonString);
+
+        try {
+            Response response = restClient.performRequest(request);
+            if (response == null) {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.CLEAR_INDEX_DATA_FAILED,
+                        "POST " + endpoint + " response null");
+            }
+            // todo: if the index doesn't exist, the response status code is 200?
+            if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                return;
+            } else {
+                throw new ElasticsearchConnectorException(
+                        ElasticsearchConnectorErrorCode.CLEAR_INDEX_DATA_FAILED,
+                        String.format(
+                                "POST %s response status code=%d",
+                                endpoint, response.getStatusLine().getStatusCode()));
+            }
+        } catch (IOException ex) {
+            throw new ElasticsearchConnectorException(
+                    ElasticsearchConnectorErrorCode.CLEAR_INDEX_DATA_FAILED, ex);
+        }
+    }
+
     /**
      * get es field name and type mapping realtion
      *

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/exception/ElasticsearchConnectorErrorCode.java

@@ -28,7 +28,8 @@ public enum ElasticsearchConnectorErrorCode implements SeaTunnelErrorCode {
     LIST_INDEX_FAILED("ELASTICSEARCH-05", "List elasticsearch index failed"),
     DROP_INDEX_FAILED("ELASTICSEARCH-06", "Drop elasticsearch index failed"),
     CREATE_INDEX_FAILED("ELASTICSEARCH-07", "Create elasticsearch index failed"),
-    ES_FIELD_TYPE_NOT_SUPPORT("ELASTICSEARCH-08", "Not support the elasticsearch field type");
+    ES_FIELD_TYPE_NOT_SUPPORT("ELASTICSEARCH-08", "Not support the elasticsearch field type"),
+    CLEAR_INDEX_DATA_FAILED("ELASTICSEARCH-09", "Clear elasticsearch index data failed");
     ;
 
     private final String code;

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java

@@ -447,35 +447,78 @@ public void tearDown() {
     }
 
     @Test
-    public void testCatalog() {
+    public void testCatalog() throws InterruptedException, JsonProcessingException {
         Map<String, Object> configMap = new HashMap<>();
         configMap.put("username", "elastic");
         configMap.put("password", "elasticsearch");
-        configMap.put("hosts", Arrays.asList("https://" + container.getHttpHostAddress()));
+        configMap.put(
+                "hosts", Collections.singletonList("https://" + container.getHttpHostAddress()));
         configMap.put("index", "st_index3");
         configMap.put("tls_verify_certificate", false);
         configMap.put("tls_verify_hostname", false);
         configMap.put("index_type", "st");
+
         final ElasticSearchCatalog elasticSearchCatalog =
                 new ElasticSearchCatalog("Elasticsearch", "", ReadonlyConfig.fromMap(configMap));
         elasticSearchCatalog.open();
+
         TablePath tablePath = TablePath.of("", "st_index3");
-        // index exists
+
+        // Verify index does not exist initially
         final boolean existsBefore = elasticSearchCatalog.tableExists(tablePath);
-        Assertions.assertFalse(existsBefore);
-        // create index
+        Assertions.assertFalse(existsBefore, "Index should not exist initially");
+
+        // Create index
         elasticSearchCatalog.createTable(tablePath, null, false);
         final boolean existsAfter = elasticSearchCatalog.tableExists(tablePath);
-        Assertions.assertTrue(existsAfter);
-        // data exists?
-        final boolean existsData = elasticSearchCatalog.isExistsData(tablePath);
-        Assertions.assertFalse(existsData);
-        // truncate
+        Assertions.assertTrue(existsAfter, "Index should be created");
+
+        // Generate and add multiple records
+        List<String> data = generateTestData();
+        StringBuilder requestBody = new StringBuilder();
+        String indexHeader = "{\"index\":{\"_index\":\"st_index3\"}}\n";
+        for (String record : data) {
+            requestBody.append(indexHeader);
+            requestBody.append(record);
+            requestBody.append("\n");
+        }
+        esRestClient.bulk(requestBody.toString());
+        Thread.sleep(2000); // Wait for data to be indexed
+
+        // Verify data exists
+        List<String> sourceFields = Arrays.asList("field1", "field2");
+        Map<String, Object> query = new HashMap<>();
+        query.put("match_all", new HashMap<>());
+        ScrollResult scrollResult =
+                esRestClient.searchByScroll("st_index3", sourceFields, query, "1m", 100);
+        Assertions.assertFalse(scrollResult.getDocs().isEmpty(), "Data should exist in the index");
+
+        // Truncate the table
         elasticSearchCatalog.truncateTable(tablePath, false);
-        Assertions.assertTrue(elasticSearchCatalog.tableExists(tablePath));
-        // drop
+        Thread.sleep(2000); // Wait for data to be indexed
+
+        // Verify data is deleted
+        scrollResult = esRestClient.searchByScroll("st_index3", sourceFields, query, "1m", 100);
+        Assertions.assertTrue(
+                scrollResult.getDocs().isEmpty(), "Data should be deleted from the index");
+
+        // Drop the table
         elasticSearchCatalog.dropTable(tablePath, false);
-        Assertions.assertFalse(elasticSearchCatalog.tableExists(tablePath));
+        Assertions.assertFalse(
+                elasticSearchCatalog.tableExists(tablePath), "Index should be dropped");
+
         elasticSearchCatalog.close();
     }
+
+    private List<String> generateTestData() throws JsonProcessingException {
+        List<String> data = new ArrayList<>();
+        ObjectMapper objectMapper = new ObjectMapper();
+        for (int i = 0; i < 10; i++) {
+            Map<String, Object> record = new HashMap<>();
+            record.put("field1", "value" + i);
+            record.put("field2", i);
+            data.add(objectMapper.writeValueAsString(record));
+        }
+        return data;
+    }
 }