review comments

Author: ywelsch

server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

@@ -137,7 +137,12 @@ private void handlePrimaryResult(final PrimaryResultT primaryResult) {
 
             @Override
             public void onResponse(Void aVoid) {
-                updateCheckPoints(primary.routingEntry(), primary::localCheckpoint, primary::globalCheckpoint);
+                successfulShards.incrementAndGet();
+                try {
+                    updateCheckPoints(primary.routingEntry(), primary::localCheckpoint, primary::globalCheckpoint);
+                } finally {
+                    decPendingAndFinishIfNeeded();
+                }
             }
 
             @Override
@@ -186,7 +191,12 @@ private void performOnReplica(final ShardRouting shard, final ReplicaRequest rep
             new ActionListener<>() {
                 @Override
                 public void onResponse(ReplicaResponse response) {
-                    updateCheckPoints(shard, response::localCheckpoint, response::globalCheckpoint);
+                    successfulShards.incrementAndGet();
+                    try {
+                        updateCheckPoints(shard, response::localCheckpoint, response::globalCheckpoint);
+                    } finally {
+                        decPendingAndFinishIfNeeded();
+                    }
                 }
 
                 @Override
@@ -213,7 +223,6 @@ public String toString() {
     }
 
     private void updateCheckPoints(ShardRouting shard, LongSupplier localCheckpointSupplier, LongSupplier globalCheckpointSupplier) {
-        successfulShards.incrementAndGet();
         try {
             primary.updateLocalCheckpointForShard(shard.allocationId().getId(), localCheckpointSupplier.getAsLong());
             primary.updateGlobalCheckpointForShard(shard.allocationId().getId(), globalCheckpointSupplier.getAsLong());
@@ -223,8 +232,6 @@ private void updateCheckPoints(ShardRouting shard, LongSupplier localCheckpointS
             // fail the primary but fall through and let the rest of operation processing complete
             final String message = String.format(Locale.ROOT, "primary failed updating local checkpoint for replica %s", shard);
             primary.failShard(message, e);
-        } finally {
-            decPendingAndFinishIfNeeded();
         }
     }
 

server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -344,15 +344,30 @@ public void handleException(TransportException exp) {
                 } else {
                     setPhase(replicationTask, "primary");
 
-                    final ActionListener<Response> referenceClosingListener = ActionListener.wrap(response -> {
+                    final ActionListener<Response> responseListener = ActionListener.wrap(response -> {
+                        adaptResponse(response, primaryShardReference.indexShard);
+
+                        if (syncGlobalCheckpointAfterOperation) {
+                            try {
+                                primaryShardReference.indexShard.maybeSyncGlobalCheckpoint("post-operation");
+                            } catch (final Exception e) {
+                                // only log non-closed exceptions
+                                if (ExceptionsHelper.unwrap(
+                                    e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
+                                    // intentionally swallow, a missed global checkpoint sync should not fail this operation
+                                    logger.info(
+                                        new ParameterizedMessage(
+                                            "{} failed to execute post-operation global checkpoint sync",
+                                            primaryShardReference.indexShard.shardId()), e);
+                                }
+                            }
+                        }
+
                         primaryShardReference.close(); // release shard operation lock before responding to caller
                         setPhase(replicationTask, "finished");
                         onCompletionListener.onResponse(response);
                     }, e -> handleException(primaryShardReference, e));
 
-                    final ActionListener<Response> responseListener = getResponseActionListener(referenceClosingListener,
-                        primaryShardReference.indexShard);
-
                     new ReplicationOperation<>(primaryRequest.getRequest(), primaryShardReference,
                         ActionListener.map(responseListener, result -> result.finalResponseIfSuccessful),
                         newReplicasProxy(), logger, actionName, primaryRequest.getPrimaryTerm()).execute();
@@ -375,25 +390,13 @@ public void onFailure(Exception e) {
 
     }
 
-    protected ActionListener<Response> getResponseActionListener(ActionListener<Response> referenceClosingListener, IndexShard shard) {
-
-        return ActionListener.wrap(response -> {
-            if (syncGlobalCheckpointAfterOperation) {
-                try {
-                    shard.maybeSyncGlobalCheckpoint("post-operation");
-                } catch (final Exception e) {
-                    // only log non-closed exceptions
-                    if (ExceptionsHelper.unwrap(
-                        e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
-                        // intentionally swallow, a missed global checkpoint sync should not fail this operation
-                        logger.info(
-                            new ParameterizedMessage(
-                                "{} failed to execute post-operation global checkpoint sync", shard.shardId()), e);
-                    }
-                }
-            }
-            referenceClosingListener.onResponse(response);
-        }, referenceClosingListener::onFailure);
+    // allows subclasses to adapt the response
+    protected void adaptResponse(Response response, IndexShard indexShard) {
+
+    }
+
+    protected ActionListener<Response> wrapResponseActionListener(ActionListener<Response> listener, IndexShard shard) {
+        return listener;
     }
 
     public static class PrimaryResult<ReplicaRequest extends ReplicationRequest<ReplicaRequest>,

test/framework/src/main/java/org/elasticsearch/index/replication/ESIndexLevelReplicationTestCase.java

@@ -601,16 +601,20 @@ protected ReplicationAction(Request request, ActionListener<Response> listener,
         public void execute() {
             try {
                 new ReplicationOperation<>(request, new PrimaryRef(),
-                    ActionListener.map(wrapListener(listener, getPrimaryShard()), result -> result.finalResponse),
+                    ActionListener.map(listener, result -> {
+                        adaptResponse(result.finalResponse, getPrimaryShard());
+                        return result.finalResponse;
+                    }),
                     new ReplicasRef(), logger, opType, primaryTerm)
                     .execute();
             } catch (Exception e) {
                 listener.onFailure(e);
             }
         }
 
-        protected ActionListener<Response> wrapListener(ActionListener<Response> listener, IndexShard indexShard) {
-            return listener;
+        // to be overridden by subclasses
+        protected void adaptResponse(Response response, IndexShard indexShard) {
+
         }
 
         protected IndexShard getPrimaryShard() {

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/bulk/TransportBulkShardOperationsAction.java

@@ -191,22 +191,15 @@ protected BulkShardOperationsResponse newResponseInstance(StreamInput in) throws
     }
 
     @Override
-    protected ActionListener<BulkShardOperationsResponse> getResponseActionListener(
-        ActionListener<BulkShardOperationsResponse> referenceClosingListener, IndexShard shard) {
-        ActionListener<BulkShardOperationsResponse> listener = super.getResponseActionListener(referenceClosingListener, shard);
-        return wrapListener(listener, shard);
+    protected void adaptResponse(BulkShardOperationsResponse response, IndexShard indexShard) {
+        adaptBulkShardOperationsResponse(response, indexShard);
     }
 
-    public static ActionListener<BulkShardOperationsResponse> wrapListener(ActionListener<BulkShardOperationsResponse> listener,
-                                                                           IndexShard shard) {
-
-        return ActionListener.wrap(response -> {
-            final SeqNoStats seqNoStats = shard.seqNoStats();
-            // return a fresh global checkpoint after the operations have been replicated for the shard follow task
-            response.setGlobalCheckpoint(seqNoStats.getGlobalCheckpoint());
-            response.setMaxSeqNo(seqNoStats.getMaxSeqNo());
-            listener.onResponse(response);
-        }, listener::onFailure);
+    public static void adaptBulkShardOperationsResponse(BulkShardOperationsResponse response, IndexShard indexShard) {
+        final SeqNoStats seqNoStats = indexShard.seqNoStats();
+        // return a fresh global checkpoint after the operations have been replicated for the shard follow task
+        response.setGlobalCheckpoint(seqNoStats.getGlobalCheckpoint());
+        response.setMaxSeqNo(seqNoStats.getMaxSeqNo());
     }
 
 }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardFollowTaskReplicationTests.java

@@ -15,7 +15,6 @@
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.delete.DeleteRequest;
-import org.elasticsearch.action.support.ActionTestUtils;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportWriteAction;
@@ -677,32 +676,26 @@ protected void performOnPrimary(IndexShard primary, BulkShardOperationsRequest r
             try (Releasable ignored = permitFuture.get()) {
                 ccrResult = TransportBulkShardOperationsAction.shardOperationOnPrimary(primary.shardId(), request.getHistoryUUID(),
                     request.getOperations(), request.getMaxSeqNoOfUpdatesOrDeletes(), primary, logger);
-                ActionTestUtils.assertNoFailureListener(result -> {
-                    TransportWriteActionTestHelper.performPostWriteActions(primary, request, ccrResult.location, logger);
-                    listener.onResponse(new PrimaryResult(ccrResult.replicaRequest(), ccrResult.finalResponseIfSuccessful));
-                }).onResponse(ccrResult);
+                TransportWriteActionTestHelper.performPostWriteActions(primary, request, ccrResult.location, logger);
             } catch (InterruptedException | ExecutionException | IOException e) {
                 throw new AssertionError(e);
             }
+            listener.onResponse(new PrimaryResult(ccrResult.replicaRequest(), ccrResult.finalResponseIfSuccessful));
         }
 
         @Override
-        protected ActionListener<BulkShardOperationsResponse> wrapListener(ActionListener<BulkShardOperationsResponse> listener,
-                                                                           IndexShard indexShard) {
-            return TransportBulkShardOperationsAction.wrapListener(listener, indexShard);
+        protected void adaptResponse(BulkShardOperationsResponse response, IndexShard indexShard) {
+            TransportBulkShardOperationsAction.adaptBulkShardOperationsResponse(response, indexShard);
         }
 
         @Override
         protected void performOnReplica(BulkShardOperationsRequest request, IndexShard replica) throws Exception {
-            final PlainActionFuture<Releasable> permitAcquiredFuture = new PlainActionFuture<>();
-            replica.acquireReplicaOperationPermit(getPrimaryShard().getPendingPrimaryTerm(),
-                getPrimaryShard().getLastKnownGlobalCheckpoint(), getPrimaryShard().getMaxSeqNoOfUpdatesOrDeletes(), permitAcquiredFuture,
-                ThreadPool.Names.SAME, request);
-            final Translog.Location location;
-            try (Releasable ignored = permitAcquiredFuture.actionGet()) {
-                location = TransportBulkShardOperationsAction.shardOperationOnReplica(request, replica, logger).location;
+            try (Releasable ignored = PlainActionFuture.get(f -> replica.acquireReplicaOperationPermit(
+                getPrimaryShard().getPendingPrimaryTerm(), getPrimaryShard().getLastKnownGlobalCheckpoint(),
+                getPrimaryShard().getMaxSeqNoOfUpdatesOrDeletes(), f, ThreadPool.Names.SAME, request))) {
+                Translog.Location location = TransportBulkShardOperationsAction.shardOperationOnReplica(request, replica, logger).location;
+                TransportWriteActionTestHelper.performPostWriteActions(replica, request, location, logger);
             }
-            TransportWriteActionTestHelper.performPostWriteActions(replica, request, location, logger);
         }
     }