OAK-11029: add workaround for elastic/elasticsearch-java#867

fabriziofortino · Aug 20, 2024 · 9c588eb · 9c588eb
1 parent e52d92d
commit 9c588eb
Showing 1 changed file with 14 additions and 1 deletion.
diff --git a/...va/org/apache/jackrabbit/oak/plugins/index/elastic/index/ElasticBulkProcessorHandler.java b/...va/org/apache/jackrabbit/oak/plugins/index/elastic/index/ElasticBulkProcessorHandler.java
@@ -42,7 +42,9 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Phaser;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
@@ -85,6 +87,8 @@ class ElasticBulkProcessorHandler {
 
     protected long totalOperations;
 
+    private final ScheduledExecutorService scheduler;
+
     private ElasticBulkProcessorHandler(@NotNull ElasticConnection elasticConnection,
                                         @NotNull String indexName,
                                         @NotNull ElasticIndexDefinition indexDefinition,
@@ -95,6 +99,13 @@ private ElasticBulkProcessorHandler(@NotNull ElasticConnection elasticConnection
         this.indexDefinition = indexDefinition;
         this.definitionBuilder = definitionBuilder;
         this.waitForESAcknowledgement = waitForESAcknowledgement;
+        // TODO: workaround for https://github.com/elastic/elasticsearch-java/pull/867 remove when fixed
+        this.scheduler = Executors.newScheduledThreadPool(BULK_PROCESSOR_CONCURRENCY + 1, (r) -> {
+            Thread t = Executors.defaultThreadFactory().newThread(r);
+            t.setName("oak-bulk-ingester#");
+            t.setDaemon(true);
+            return t;
+        });
         this.bulkIngester = initBulkIngester();
     }
 
@@ -150,7 +161,7 @@ private BulkIngester<String> initBulkIngester() {
             if (indexDefinition.bulkFlushIntervalMs > 0) {
                 b = b.flushInterval(indexDefinition.bulkFlushIntervalMs, TimeUnit.MILLISECONDS);
             }
-            return b.maxConcurrentRequests(BULK_PROCESSOR_CONCURRENCY);
+            return b.scheduler(scheduler).maxConcurrentRequests(BULK_PROCESSOR_CONCURRENCY);
         });
     }
 
@@ -206,6 +217,8 @@ public boolean close() throws IOException {
             }
         }
 
+        scheduler.shutdownNow();
+
         checkFailures();
 
         if (LOG.isTraceEnabled()) {