CCR should retry on CircuitBreakingException (#62013)

CCR shard follow task can hit CircuitBreakingException on the leader
cluster (read changes requests) or the follower cluster (bulk requests).
CCR should retry on CircuitBreakingException as it's a transient error.

Author: dnhatn

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.action.UnavailableShardsException;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.transport.NetworkExceptionHelper;
 import org.elasticsearch.common.unit.TimeValue;
@@ -524,7 +525,8 @@ static boolean shouldRetry(final Exception e) {
             actual instanceof NodeClosedException ||
             actual instanceof NoSuchRemoteClusterException ||
             (actual.getMessage() != null && actual.getMessage().contains("TransportService is closed")) ||
-            actual instanceof EsRejectedExecutionException;
+            actual instanceof EsRejectedExecutionException ||
+            actual instanceof CircuitBreakingException;
     }
 
     // These methods are protected for testing purposes:

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

@@ -7,9 +7,11 @@
 
 import org.elasticsearch.action.UnavailableShardsException;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.index.seqno.LocalCheckpointTracker;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
@@ -285,7 +287,10 @@ private static TestRun createTestRun(long startSeqNo, long startMappingVersion,
                     List<TestResponse> item = new ArrayList<>();
                     // Sometimes add a random retryable error
                     if (sometimes()) {
-                        Exception error = new UnavailableShardsException(new ShardId("test", "test", 0), "");
+                        Exception error = randomFrom(
+                            new UnavailableShardsException(new ShardId("test", "test", 0), ""),
+                            new CircuitBreakingException("test", randomInt(), randomInt()),
+                            new EsRejectedExecutionException("test"));
                         item.add(new TestResponse(error, mappingVersion, settingsVersion, null));
                     }
                     // Sometimes add an empty shard changes response to also simulate a leader shard lagging behind