feat(elasticsearch): allow bulk delete (#8424)

Author: david-leifker

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/update/ESBulkProcessor.java

@@ -47,6 +47,9 @@ public static ESBulkProcessor.ESBulkProcessorBuilder builder(RestHighLevelClient
     @NonNull
     private Boolean async = false;
     @Builder.Default
+    @NonNull
+    private Boolean batchDelete = false;
+    @Builder.Default
     private Integer bulkRequestsLimit = 500;
     @Builder.Default
     private Integer bulkFlushPeriod = 1;
@@ -62,12 +65,13 @@ public static ESBulkProcessor.ESBulkProcessorBuilder builder(RestHighLevelClient
     @Getter(AccessLevel.NONE)
     private final BulkProcessor bulkProcessor;
 
-    private ESBulkProcessor(@NonNull RestHighLevelClient searchClient, @NonNull Boolean async, Integer bulkRequestsLimit,
-                            Integer bulkFlushPeriod, Integer numRetries, Long retryInterval,
+    private ESBulkProcessor(@NonNull RestHighLevelClient searchClient, @NonNull Boolean async, @NonNull Boolean batchDelete,
+                            Integer bulkRequestsLimit, Integer bulkFlushPeriod, Integer numRetries, Long retryInterval,
                             TimeValue defaultTimeout, WriteRequest.RefreshPolicy writeRequestRefreshPolicy,
                             BulkProcessor ignored) {
         this.searchClient = searchClient;
         this.async = async;
+        this.batchDelete = batchDelete;
         this.bulkRequestsLimit = bulkRequestsLimit;
         this.bulkFlushPeriod = bulkFlushPeriod;
         this.numRetries = numRetries;
@@ -103,8 +107,10 @@ public Optional<BulkByScrollResponse> deleteByQuery(QueryBuilder queryBuilder, b
         deleteByQueryRequest.indices(indices);
 
         try {
-            // flush pending writes
-            bulkProcessor.flush();
+            if (!batchDelete) {
+                // flush pending writes
+                bulkProcessor.flush();
+            }
             // perform delete after local flush
             final BulkByScrollResponse deleteResponse = searchClient.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
             MetricUtils.counter(this.getClass(), ES_WRITES_METRIC).inc(deleteResponse.getTotal());

metadata-service/configuration/src/main/resources/application.yml

@@ -168,6 +168,7 @@ elasticsearch:
     numRetries: ${ES_BULK_NUM_RETRIES:3}
     retryInterval: ${ES_BULK_RETRY_INTERVAL:1}
     refreshPolicy: ${ES_BULK_REFRESH_POLICY:NONE}
+    enableBatchDelete: ${ES_BULK_ENABLE_BATCH_DELETE:false}
   index:
     prefix: ${INDEX_PREFIX:}
     numShards: ${ELASTICSEARCH_NUM_SHARDS_PER_INDEX:1}

metadata-service/factories/src/main/java/com/linkedin/gms/factory/search/ElasticSearchBulkProcessorFactory.java

@@ -41,6 +41,9 @@ public class ElasticSearchBulkProcessorFactory {
   @Value("#{new Boolean('${elasticsearch.bulkProcessor.async}')}")
   private boolean async;
 
+  @Value("#{new Boolean('${elasticsearch.bulkProcessor.enableBatchDelete}')}")
+  private boolean enableBatchDelete;
+
   @Value("${elasticsearch.bulkProcessor.refreshPolicy}")
   private String refreshPolicy;
 
@@ -53,6 +56,7 @@ protected ESBulkProcessor getInstance() {
             .bulkRequestsLimit(bulkRequestsLimit)
             .retryInterval(retryInterval)
             .numRetries(numRetries)
+            .batchDelete(enableBatchDelete)
             .writeRequestRefreshPolicy(WriteRequest.RefreshPolicy.valueOf(refreshPolicy))
             .build();
   }