henning’s comment

Author: dnhatn

server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -415,16 +415,16 @@ void phase1(IndexCommit snapshot, long globalCheckpoint, IntSupplier translogOps
                     phase1FileNames.size(), new ByteSizeValue(totalSizeInBytes),
                     phase1ExistingFileNames.size(), new ByteSizeValue(existingTotalSizeInBytes));
                 final StepListener<Void> sendFileInfoStep = new StepListener<>();
-                final StepListener<Void> sendFileChunkStep = new StepListener<>();
+                final StepListener<Void> sendFilesStep = new StepListener<>();
                 final StepListener<Void> cleanFilesStep = new StepListener<>();
                 cancellableThreads.execute(() ->
                     recoveryTarget.receiveFileInfo(phase1FileNames, phase1FileSizes, phase1ExistingFileNames,
                         phase1ExistingFileSizes, translogOps.getAsInt(), sendFileInfoStep));
 
                 sendFileInfoStep.whenComplete(r ->
-                    sendFiles(store, phase1Files.toArray(new StoreFileMetaData[0]), translogOps, sendFileChunkStep), listener::onFailure);
+                    sendFiles(store, phase1Files.toArray(new StoreFileMetaData[0]), translogOps, sendFilesStep), listener::onFailure);
 
-                sendFileChunkStep.whenComplete(r ->
+                sendFilesStep.whenComplete(r ->
                     cleanFiles(store, recoverySourceMetadata, translogOps, globalCheckpoint, cleanFilesStep), listener::onFailure);
 
                 final long totalSize = totalSizeInBytes;
@@ -726,10 +726,20 @@ protected void doRun() throws Exception {
             assert Transports.assertNotTransportThread(RecoverySourceHandler.this + "[send file chunk]");
             while (true) {
                 assert semaphore.availablePermits() == 0;
+                cancellableThreads.checkForCancel();
                 if (error.get() != null) {
                     handleErrorOnSendFiles(store, error.get().v2(), new StoreFileMetaData[]{error.get().v1()});
                     throw error.get().v2();
                 }
+                if (canSendMore() == false) {
+                    semaphore.release();
+                    // Here we have to retry before abort to avoid a race situation where the other threads have flipped `canSendMore`
+                    // condition but they are not going to resume the sending process because this thread still holds the semaphore.
+                    final boolean changed = canSendMore() || error.get() != null;
+                    if (changed == false || semaphore.tryAcquire() == false) {
+                        break;
+                    }
+                }
                 final FileChunk chunk = readNextChunk();
                 if (chunk == null) {
                     semaphore.release(); // allow other threads respond if we are not done yet.
@@ -757,15 +767,6 @@ protected void doRun() throws Exception {
                             })
                     )
                 );
-                if (canSendMore() == false) {
-                    semaphore.release();
-                    // Here we have to retry before abort to avoid a race situation where the other threads have flipped `canSendMore`
-                    // condition but they are not going to resume the sending process because this thread still holds the semaphore.
-                    final boolean changed = canSendMore() || error.get() != null;
-                    if (changed == false || semaphore.tryAcquire() == false) {
-                        break;
-                    }
-                }
             }
         }