Fix

Tim-Brooks · Sep 11, 2024 · b0e95ba · b0e95ba
1 parent 50965e1
commit b0e95ba
Showing 1 changed file with 4 additions and 2 deletions.
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/IncrementalBulkService.java b/server/src/main/java/org/elasticsearch/action/bulk/IncrementalBulkService.java
@@ -140,6 +140,7 @@ public void addItems(List<DocWriteRequest<?>> items, Releasable releasable, Runn
                         try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
                             requestContext.restore();
                             final ArrayList<Releasable> toRelease = new ArrayList<>(releasables);
+                            releasables.clear();
                             client.bulk(bulkRequest, ActionListener.runAfter(new ActionListener<>() {
 
                                 @Override
@@ -160,7 +161,6 @@ public void onFailure(Exception e) {
                                 nextItems.run();
                             }));
                         }
-                        releasables.clear();
                     } else {
                         nextItems.run();
                     }
@@ -184,6 +184,7 @@ public void lastItems(List<DocWriteRequest<?>> items, Releasable releasable, Act
                     try (ThreadContext.StoredContext ignored = threadContext.stashContext()) {
                         requestContext.restore();
                         final ArrayList<Releasable> toRelease = new ArrayList<>(releasables);
+                        releasables.clear();
                         client.bulk(bulkRequest, ActionListener.runAfter(new ActionListener<>() {
 
                             private final boolean isFirstRequest = incrementalRequestSubmitted == false;
@@ -201,7 +202,6 @@ public void onFailure(Exception e) {
                             }
                         }, () -> toRelease.forEach(Releasable::close)));
                     }
-                    releasables.clear();
                 } else {
                     errorResponse(listener);
                 }
@@ -270,6 +270,8 @@ private boolean internalAddItems(List<DocWriteRequest<?>> items, Releasable rele
                 return true;
             } catch (EsRejectedExecutionException e) {
                 handleBulkFailure(incrementalRequestSubmitted == false, e);
+                releasables.forEach(Releasable::close);
+                releasables.clear();
                 return false;
             }
         }