Upgrade Elasticsearch 8.x to version 8.15.0. [DOX-412]

Incompatible change in elastic/elasticsearch-java#830. (Totally meaningless release notes entry: "Fixed bug in BulkIngester") https://www.elastic.co/guide/en/elasticsearch/client/java-api-client/8.15/release-highlights.html
hbz · Aug 12, 2024 · a49f1c4 · a49f1c4
1 parent ca09bd4
commit a49f1c4
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/settings.gradle b/settings.gradle
@@ -14,7 +14,7 @@ gradle.rootProject {
             'commons-cli'    : '1.3.1',
             'commons-io'     : '2.7',
             'elasticsearch2' : '2.2.1',
-            'elasticsearch8' : '8.14.1',
+            'elasticsearch8' : '8.15.0',
             'htsjdk'         : '4.0.1',
             'jackson'        : '2.13.4.2',
             'jdk'            : '17',

diff --git a/src/main/java/hbz/limetrans/ElasticsearchClientV8.java b/src/main/java/hbz/limetrans/ElasticsearchClientV8.java
@@ -45,6 +45,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -64,6 +67,7 @@ public class ElasticsearchClientV8 extends ElasticsearchClient { // checkstyle-d
 
     private BulkIngester<Void> mBulkIngester;
     private RestClientTransport mTransport;
+    private ScheduledExecutorService mBulkScheduler;
     private co.elastic.clients.elasticsearch.ElasticsearchClient mClient;
 
     public ElasticsearchClientV8(final Settings aSettings) {
@@ -75,6 +79,7 @@ public ElasticsearchClientV8(final Settings aSettings) {
     @Override
     public void reset() {
         mBulkIngester = null;
+        mBulkScheduler = null;
 
         super.reset();
     }
@@ -269,8 +274,16 @@ private void addBulk(final Function<BulkOperation.Builder, ObjectBuilder<BulkOpe
 
     @Override
     protected void createBulk(final int aBulkActions, final int aBulkRequests) {
+        mBulkScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            final Thread t = Executors.defaultThreadFactory().newThread(r);
+            t.setName("limetrans-bulk-ingester-executor#");
+            t.setDaemon(true);
+            return t;
+        });
+
         mBulkIngester = BulkIngester.of(b -> b
                 .client(mClient)
+                .scheduler(mBulkScheduler)
                 .listener(new ElasticsearchBulkListener(this))
                 .maxOperations(aBulkActions)
                 .maxSize(mBulkSizeValue)
@@ -287,10 +300,13 @@ protected boolean isBulkClosed() {
     protected boolean closeBulk() throws InterruptedException {
         try {
             mBulkIngester.close();
+            mBulkScheduler.shutdown();
+            mBulkScheduler.awaitTermination(2, TimeUnit.MINUTES);
             return mBulkIngester.pendingRequests() == 0;
         }
         finally {
             mBulkIngester = null;
+            mBulkScheduler = null;
         }
     }