Fix failing CCR retention lease test

Finally! This commit should fix the issues with the CCR retention lease
that has been plaguing build failures. The issue here is that we are
trying to prevent the clear session requests from being executed until
after we have been able to validate that retention leases are being
renewed. However, we were only blocking the clear session requests but
not blocking them when they are proxied through another node. This
commit addresses that.

Closes #39268

Author: jasontedor

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

@@ -40,9 +40,9 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.snapshots.RestoreInfo;
 import org.elasticsearch.snapshots.RestoreService;
-import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.transport.ConnectTransportException;
+import org.elasticsearch.transport.TransportActionProxy;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.CcrIntegTestCase;
 import org.elasticsearch.xpack.ccr.action.repositories.ClearCcrRestoreSessionAction;
@@ -191,7 +191,6 @@ public void testRetentionLeaseIsTakenAtTheStartOfRecovery() throws Exception {
 
     }
 
-    @TestLogging(value = "org.elasticsearch.xpack.ccr:trace")
     public void testRetentionLeaseIsRenewedDuringRecovery() throws Exception {
         final String leaderIndex = "leader";
         final int numberOfShards = randomIntBetween(1, 3);
@@ -212,10 +211,9 @@ public void testRetentionLeaseIsRenewedDuringRecovery() throws Exception {
                     (MockTransportService) getFollowerCluster().getInstance(TransportService.class, senderNode.value.getName());
             senderTransportService.addSendBehavior(
                     (connection, requestId, action, request, options) -> {
-                        logger.info("action [{}]", action);
-                        if (ClearCcrRestoreSessionAction.NAME.equals(action)) {
+                        if (ClearCcrRestoreSessionAction.NAME.equals(action)
+                                || TransportActionProxy.getProxyAction(ClearCcrRestoreSessionAction.NAME).equals(action)) {
                             try {
-                                logger.info("latching");
                                 latch.await();
                             } catch (final InterruptedException e) {
                                 fail(e.toString());
@@ -436,7 +434,8 @@ public void testUnfollowRemovesRetentionLeases() throws Exception {
                         (MockTransportService) getFollowerCluster().getInstance(TransportService.class, senderNode.value.getName());
                 senderTransportService.addSendBehavior(
                         (connection, requestId, action, request, options) -> {
-                            if (RetentionLeaseActions.Remove.ACTION_NAME.equals(action)) {
+                            if (RetentionLeaseActions.Remove.ACTION_NAME.equals(action)
+                                    || TransportActionProxy.getProxyAction(RetentionLeaseActions.Remove.ACTION_NAME).equals(action)) {
                                 final RetentionLeaseActions.RemoveRequest removeRequest = (RetentionLeaseActions.RemoveRequest) request;
                                 if (shardIds.contains(removeRequest.getShardId().id())) {
                                     final String primaryShardNodeId =
@@ -520,7 +519,8 @@ public void testUnfollowFailsToRemoveRetentionLeases() throws Exception {
                         (MockTransportService) getFollowerCluster().getInstance(TransportService.class, senderNode.value.getName());
                 senderTransportService.addSendBehavior(
                         (connection, requestId, action, request, options) -> {
-                            if (RetentionLeaseActions.Remove.ACTION_NAME.equals(action)) {
+                            if (RetentionLeaseActions.Remove.ACTION_NAME.equals(action)
+                                    || TransportActionProxy.getProxyAction(RetentionLeaseActions.Remove.ACTION_NAME).equals(action)) {
                                 final RetentionLeaseActions.RemoveRequest removeRequest = (RetentionLeaseActions.RemoveRequest) request;
                                 if (shardIds.contains(removeRequest.getShardId().id())) {
                                     throw randomBoolean()