Use cleanupOnly listener instead

Author: fcofdez

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

@@ -333,43 +333,23 @@ public void onFailure(Exception e) {
         if (indexShard.routingEntry().isSearchable() == false && recoveryState.getPrimary()) {
             assert preExistingRequest == null;
             assert indexShard.indexSettings().getIndexMetadata().isSearchableSnapshot() == false;
-            try (onCompletion) {
-                client.execute(
-                    StatelessPrimaryRelocationAction.TYPE,
-                    new StatelessPrimaryRelocationAction.Request(
-                        recoveryId,
-                        indexShard.shardId(),
-                        transportService.getLocalNode(),
-                        indexShard.routingEntry().allocationId().getId(),
-                        recoveryTarget.clusterStateVersion()
-                    ),
-                    new ActionListener<>() {
-                        @Override
-                        public void onResponse(ActionResponse.Empty ignored) {
-                            onGoingRecoveries.markRecoveryAsDone(recoveryId);
-                        }
-
-                        @Override
-                        public void onFailure(Exception e) {
-                            final var cause = ExceptionsHelper.unwrapCause(e);
-                            final var sendShardFailure =
-                                // these indicate the source shard has already failed, which will independently notify the master and fail
-                                // the target shard
-                                false == (cause instanceof ShardNotFoundException
-                                    || cause instanceof IndexNotFoundException
-                                    || cause instanceof AlreadyClosedException);
-
-                            // TODO retries? See RecoveryResponseHandler#handleException
-                            onGoingRecoveries.failRecovery(
-                                recoveryId,
-                                new RecoveryFailedException(recoveryState, null, e),
-                                sendShardFailure
-                            );
-                        }
-                    }
-                );
-                return;
-            }
+            client.execute(
+                StatelessPrimaryRelocationAction.TYPE,
+                new StatelessPrimaryRelocationAction.Request(
+                    recoveryId,
+                    indexShard.shardId(),
+                    transportService.getLocalNode(),
+                    indexShard.routingEntry().allocationId().getId(),
+                    recoveryTarget.clusterStateVersion()
+                ),
+                cleanupOnly.delegateFailure((l, unused) -> {
+                    ActionListener.completeWith(l, () -> {
+                        onGoingRecoveries.markRecoveryAsDone(recoveryId);
+                        return null;
+                    });
+                })
+            );
+            return;
         }
 
         record StartRecoveryRequestToSend(StartRecoveryRequest startRecoveryRequest, String actionName, TransportRequest requestToSend) {}