Fix enforcing that only one recovery gets retried and succeeds (#91845)

We have to register an explicit handler for an PeerRecoveryTargetService.Actions.FINALIZE event in order to finalizeReceived to be set.

Fixes #91087
Fixes #91041

server/src/internalClusterTest/java/org/elasticsearch/indices/recovery/IndexRecoveryIT.java

@@ -798,7 +798,6 @@ public class IndexRecoveryIT extends AbstractIndexRecoveryIntegTestCase {
         assertThat(indexState.recoveredBytesPercent(), lessThanOrEqualTo(100.0f));
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/91087")
     public void testTransientErrorsDuringRecoveryAreRetried() throws Exception {
         final String recoveryActionToBlock = randomFrom(
             PeerRecoveryTargetService.Actions.PREPARE_TRANSLOG,

test/framework/src/main/java/org/elasticsearch/indices/recovery/AbstractIndexRecoveryIntegTestCase.java

@@ -178,12 +178,11 @@ public abstract class AbstractIndexRecoveryIntegTestCase extends ESIntegTestCase
                 redTransportService.disconnectFromNode(blueTransportService.getLocalDiscoNode());
             }
         };
-        TransientReceiveRejected handlingBehavior = new TransientReceiveRejected(
-            recoveryActionToBlock,
-            recoveryStarted,
-            finalizeReceived,
-            connectionBreaker
-        );
+        TransientReceiveRejected handlingBehavior = new TransientReceiveRejected(recoveryActionToBlock, recoveryStarted, connectionBreaker);
+        redTransportService.addRequestHandlingBehavior(PeerRecoveryTargetService.Actions.FINALIZE, (handler, request, channel, task) -> {
+            finalizeReceived.set(true);
+            handler.messageReceived(request, channel, task);
+        });
         redTransportService.addRequestHandlingBehavior(recoveryActionToBlock, handlingBehavior);
 
         try {
@@ -580,19 +579,12 @@ public abstract class AbstractIndexRecoveryIntegTestCase extends ESIntegTestCase
 
         private final String actionName;
         private final AtomicBoolean recoveryStarted;
-        private final AtomicBoolean finalizeReceived;
         private final Runnable connectionBreaker;
         private final AtomicInteger blocksRemaining;
 
-        private TransientReceiveRejected(
-            String actionName,
-            AtomicBoolean recoveryStarted,
-            AtomicBoolean finalizeReceived,
-            Runnable connectionBreaker
-        ) {
+        private TransientReceiveRejected(String actionName, AtomicBoolean recoveryStarted, Runnable connectionBreaker) {
             this.actionName = actionName;
             this.recoveryStarted = recoveryStarted;
-            this.finalizeReceived = finalizeReceived;
             this.connectionBreaker = connectionBreaker;
             this.blocksRemaining = new AtomicInteger(randomIntBetween(1, 3));
         }
@@ -605,9 +597,6 @@ public abstract class AbstractIndexRecoveryIntegTestCase extends ESIntegTestCase
             Task task
         ) throws Exception {
             recoveryStarted.set(true);
-            if (actionName.equals(PeerRecoveryTargetService.Actions.FINALIZE)) {
-                finalizeReceived.set(true);
-            }
             if (blocksRemaining.getAndUpdate(i -> i == 0 ? 0 : i - 1) != 0) {
                 String rejected = "rejected";
                 String circuit = "circuit";

x-pack/plugin/snapshot-based-recoveries/src/internalClusterTest/java/org/elasticsearch/xpack/snapshotbasedrecoveries/recovery/IndexRecoveryWithSnapshotsIT.java

@@ -22,7 +22,6 @@ public class IndexRecoveryWithSnapshotsIT extends AbstractIndexRecoveryIntegTest
         return CollectionUtils.appendToCopy(super.nodePlugins(), ConfigurableMockSnapshotBasedRecoveriesPlugin.class);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/91087")
     public void testTransientErrorsDuringRecoveryAreRetried() throws Exception {
         checkTransientErrorsDuringRecoveryAreRetried(PeerRecoveryTargetService.Actions.RESTORE_FILE_FROM_SNAPSHOT);
     }