bug

Author: tlrx

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

@@ -692,8 +692,10 @@ private final class AsyncReplicaAction extends AbstractRunnable implements Actio
         public void onResponse(Releasable releasable) {
             assert replica.getActiveOperationsCount() != 0 : "must perform shard operation under a permit";
             try {
-                shardOperationOnReplica(replicaRequest.getRequest(), replica, ActionListener.wrap((replicaResult) -> {
-                    replicaResult.runPostReplicaActions(new ReplicaPostReplicationActionsListener() {
+                shardOperationOnReplica(
+                    replicaRequest.getRequest(),
+                    replica,
+                    ActionListener.wrap((replicaResult) -> replicaResult.runPostReplicaActions(new ReplicaPostReplicationActionsListener() {
                         @Override
                         public void onResponse(long globalCheckpoint, long localCheckpoint) {
                             final ReplicaResponse response = new ReplicaResponse(localCheckpoint, globalCheckpoint);
@@ -716,11 +718,11 @@ public void onFailure(Exception e) {
                             Releasables.closeWhileHandlingException(releasable);
                             responseWithFailure(e);
                         }
-                    });
-                }, e -> {
-                    Releasables.closeWhileHandlingException(releasable); // release shard operation lock before responding to caller
-                    AsyncReplicaAction.this.onFailure(e);
-                }));
+                    }), e -> {
+                        Releasables.closeWhileHandlingException(releasable); // release shard operation lock before responding to caller
+                        AsyncReplicaAction.this.onFailure(e);
+                    })
+                );
                 // TODO: Evaluate if we still need to catch this exception
             } catch (Exception e) {
                 Releasables.closeWhileHandlingException(releasable); // release shard operation lock before responding to caller

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

@@ -511,8 +511,13 @@ private void maybeFinish() {
         }
 
         private void updateCheckpoints() {
-            this.globalCheckpoint.accumulateAndGet(indexShard.getLastSyncedGlobalCheckpoint(), Math::max);
-            this.localCheckpoint.accumulateAndGet(indexShard.getLocalCheckpoint(), Math::max);
+            try {
+                this.globalCheckpoint.accumulateAndGet(indexShard.getLastSyncedGlobalCheckpoint(), Math::max);
+                this.localCheckpoint.accumulateAndGet(indexShard.getLocalCheckpoint(), Math::max);
+            } catch (Exception e) {
+                logger.warn("Failed to retrieve checkpoints", e);
+                assert false : e;
+            }
         }
 
         void run() {
@@ -559,8 +564,11 @@ public void onFailure(Exception e) {
             if (sync) {
                 assert pendingOps.get() > 0;
                 indexShard.syncAfterWrite(location, e -> {
-                    updateCheckpoints();
-                    syncFailure.set(e);
+                    if (e != null) {
+                        syncFailure.set(e);
+                    } else {
+                        updateCheckpoints();
+                    }
                     maybeFinish();
                 });
                 updateCheckpoints();