More

Author: Tim-Brooks

server/src/main/java/org/elasticsearch/action/bulk/IncrementalBulkService.java

@@ -91,19 +91,18 @@ public static class Handler implements Releasable {
         public static final BulkRequest.IncrementalState EMPTY_STATE = new BulkRequest.IncrementalState(Collections.emptyMap(), true);
 
         private final Client client;
-        private final IndexingPressure indexingPressure;
         private final ActiveShardCount waitForActiveShards;
         private final TimeValue timeout;
         private final String refresh;
 
         private final ArrayList<Releasable> releasables = new ArrayList<>(4);
         private final ArrayList<BulkResponse> responses = new ArrayList<>(2);
+        private final IndexingPressure.Coordinating coordinatingOperation;
         private boolean closed = false;
         private boolean globalFailure = false;
         private boolean incrementalRequestSubmitted = false;
         private boolean bulkInProgress = false;
         private Exception bulkActionLevelFailure = null;
-        private IndexingPressure.Coordinating coordinatingOperation;
         private BulkRequest bulkRequest = null;
 
         protected Handler(
@@ -114,10 +113,10 @@ protected Handler(
             @Nullable String refresh
         ) {
             this.client = client;
-            this.indexingPressure = indexingPressure;
             this.waitForActiveShards = waitForActiveShards != null ? ActiveShardCount.parseString(waitForActiveShards) : null;
             this.timeout = timeout;
             this.refresh = refresh;
+            this.coordinatingOperation = indexingPressure.markCoordinatingOperationStarted(0, 0, false);
             createNewBulkRequest(EMPTY_STATE);
         }
 
@@ -152,8 +151,7 @@ public void onFailure(Exception e) {
                             }
                         }, () -> {
                             bulkInProgress = false;
-                            coordinatingOperation.close();
-                            coordinatingOperation = null;
+                            coordinatingOperation.releaseCurrent();
                             toRelease.forEach(Releasable::close);
                             nextItems.run();
                         }));
@@ -194,8 +192,7 @@ public void onFailure(Exception e) {
                             errorResponse(listener);
                         }
                     }, () -> {
-                        coordinatingOperation.close();
-                        coordinatingOperation = null;
+                        coordinatingOperation.releaseCurrent();
                         toRelease.forEach(Releasable::close);
                     }));
                 } else {
@@ -208,13 +205,13 @@ public void onFailure(Exception e) {
         public void close() {
             closed = true;
             coordinatingOperation.close();
-            coordinatingOperation = null;
             releasables.forEach(Releasable::close);
             releasables.clear();
         }
 
         private void shortCircuitDueToTopLevelFailure(List<DocWriteRequest<?>> items, Releasable releasable) {
             assert releasables.isEmpty();
+            assert coordinatingOperation.currentSize() == 0;
             assert bulkRequest == null;
             if (globalFailure == false) {
                 addItemLevelFailures(items);
@@ -263,8 +260,7 @@ private boolean internalAddItems(List<DocWriteRequest<?>> items, Releasable rele
                 return true;
             } catch (EsRejectedExecutionException e) {
                 handleBulkFailure(incrementalRequestSubmitted == false, e);
-                coordinatingOperation.close();
-                coordinatingOperation = null;
+                coordinatingOperation.releaseCurrent();
                 releasables.forEach(Releasable::close);
                 releasables.clear();
                 return false;
@@ -275,7 +271,6 @@ private void createNewBulkRequest(BulkRequest.IncrementalState incrementalState)
             assert bulkRequest == null;
             bulkRequest = new BulkRequest();
             bulkRequest.incrementalState(incrementalState);
-            coordinatingOperation = indexingPressure.markCoordinatingOperationStarted(0, 0, false);
 
             if (waitForActiveShards != null) {
                 bulkRequest.waitForActiveShards(waitForActiveShards);

server/src/main/java/org/elasticsearch/index/IndexingPressure.java

@@ -146,16 +146,17 @@ public Coordinating markCoordinatingOperationStarted(int operations, long bytes,
 
     public class Coordinating implements Releasable {
 
-        private final AtomicBoolean called = new AtomicBoolean();
+        private final AtomicBoolean closed = new AtomicBoolean();
         private final boolean forceExecution;
-        private int coordinatingOperations = 0;
+        private int currentOperations = 0;
         private long currentSize = 0;
 
         public Coordinating(boolean forceExecution) {
             this.forceExecution = forceExecution;
         }
 
         public void increment(int operations, long bytes) {
+            assert closed.get() == false;
             long combinedBytes = currentCombinedCoordinatingAndPrimaryBytes.addAndGet(bytes);
             long replicaWriteBytes = currentReplicaBytes.get();
             long totalBytes = combinedBytes + replicaWriteBytes;
@@ -184,7 +185,7 @@ public void increment(int operations, long bytes) {
                     false
                 );
             }
-            coordinatingOperations += operations;
+            currentOperations += operations;
             currentSize += bytes;
             logger.trace(() -> Strings.format("adding [%d] coordinating operations and [%d] bytes", operations, bytes));
             currentCoordinatingBytes.getAndAdd(bytes);
@@ -195,6 +196,18 @@ public void increment(int operations, long bytes) {
             totalCoordinatingRequests.getAndIncrement();
         }
 
+        public long currentSize() {
+            return currentSize;
+        }
+
+        public void releaseCurrent() {
+            currentCombinedCoordinatingAndPrimaryBytes.getAndAdd(-currentSize);
+            currentCoordinatingBytes.getAndAdd(-currentSize);
+            currentCoordinatingOps.getAndAdd(-currentOperations);
+            currentSize = 0;
+            currentOperations = 0;
+        }
+
         public boolean shouldSplit() {
             long currentUsage = (currentCombinedCoordinatingAndPrimaryBytes.get() + currentReplicaBytes.get());
             if (currentUsage >= highWatermark && currentSize >= highWatermarkSize) {
@@ -217,13 +230,9 @@ public boolean shouldSplit() {
 
         @Override
         public void close() {
-            if (called.compareAndSet(false, true)) {
-                logger.trace(
-                    () -> Strings.format("removing [%d] coordinating operations and [%d] bytes", coordinatingOperations, currentSize)
-                );
-                currentCombinedCoordinatingAndPrimaryBytes.getAndAdd(-currentSize);
-                currentCoordinatingBytes.getAndAdd(-currentSize);
-                currentCoordinatingOps.getAndAdd(-coordinatingOperations);
+            if (closed.compareAndSet(false, true)) {
+                logger.trace(() -> Strings.format("removing [%d] coordinating operations and [%d] bytes", currentOperations, currentSize));
+                releaseCurrent();
             } else {
                 logger.error("IndexingPressure memory is adjusted twice", new IllegalStateException("Releasable is called twice"));
                 assert false : "IndexingPressure is adjusted twice";