### CI Link

https://gradle-enterprise.elastic.co/s/v2am7ghdg5h7o/tests/:x-pack:plugin:async-search:internalClusterTest/org.elasticsearch.xpack.search.CrossClusterAsyncSearchIT/testCancellationViaTimeoutWithAllowPartialResultsSetToFalse?top-execution=1

### Repro line

./gradlew ':x-pack:plugin:async-search:internalClusterTest' --tests "org.elasticsearch.xpack.search.CrossClusterAsyncSearchIT.testCancellationViaTimeoutWithAllowPartialResultsSetToFalse" -Dtests.seed=B04664FE6047F39D -Dtests.locale=mk-MK -Dtests.timezone=Asia/Macao -Druntime.java=20	

### Does it reproduce?

No

### Applicable branches

main; PR branch is async-search/cancellation #97029

### Failure history

Never fails locally for me. Ran it 100 times with and without the seed indicated. All branch builds worked (4 or 5 times). Failed on first build in cloud CI after being merged to main.

### Failure excerpt

```
java.lang.AssertionError: Request breaker not reset to 0 on node: main-cluster-1 
  | Expected: <0L> |  
  | but: was <3536L> |  

at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:18) |  
-- | --
  |   | at org.junit.Assert.assertThat(Assert.java:956) |  
  |   | at org.elasticsearch.test.InternalTestCluster.lambda$ensureEstimatedStats$41(InternalTestCluster.java:2435) |  
  |   | at org.elasticsearch.test.ESTestCase.assertBusy(ESTestCase.java:1172) |  
  |   | at org.elasticsearch.test.ESTestCase.assertBusy(ESTestCase.java:1145) |  
  |   | at org.elasticsearch.test.InternalTestCluster.ensureEstimatedStats(InternalTestCluster.java:2433) |  
  |   | at org.elasticsearch.test.TestCluster.assertAfterTest(TestCluster.java:87) |  
  |   | at org.elasticsearch.test.InternalTestCluster.assertAfterTest(InternalTestCluster.java:2482) |  
  |   | at org.elasticsearch.test.AbstractMultiClustersTestCase.assertAfterTest(AbstractMultiClustersTestCase.java:131) |  
  |   | at jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104) |  
  |   | at java.lang.reflect.Method.invoke(Method.java:578) |  
  |   | at com.carrotsearch.randomizedtesting.RandomizedRunner.invoke(RandomizedRunner.java:1758) |  
  |   | at com.carrotsearch.randomizedtesting.RandomizedRunner$10.evaluate(RandomizedRunner.java:1004) |  
  |   | at com.carrotsearch.randomizedtesting.rules.StatementAdapter.evaluate(StatementAdapter.java:36) |  
  |   | at org.apache.lucene.tests.util.TestRuleSetupTeardownChained$1.evaluate(TestRuleSetupTeardownChained.java:48) |  
  |   | at org.apache.lucene.tests.util.AbstractBeforeAfterRule$1.evaluate(AbstractBeforeAfterRule.java:43) |  
  |   | at org.apache.lucene.tests.util.TestRuleThreadAndTestName$1.evaluate(TestRuleThreadAndTestName.java:45) |  
  |   | at org.apache.lucene.tests.util.TestRuleIgnoreAfterMaxFailures$1.evaluate(TestRuleIgnoreAfterMaxFailures.java:60) |  
  |   | at org.apache.lucene.tests.util.TestRuleMarkFailure$1.evaluate(TestRuleMarkFailure.java:44)
```

```
[2023-06-30T23:55:59,727][INFO ][o.e.c.m.MetadataCreateIndexService] [main-cluster-0] [.async-search] creating index, cause [auto(bulk api)], templates [], shards [1]/[0] |  
  | [2023-06-30T23:55:59,857][INFO ][o.e.c.r.a.AllocationService] [main-cluster-0] current.health="GREEN" message="Cluster health status changed from [YELLOW] to [GREEN] (reason: [shards started [[.async-search][0]]])." previous.health="YELLOW" reason="shards started [[.async-search][0]]" |  
  | јун. 30, 2023 11:55:59 PM com.carrotsearch.randomizedtesting.RandomizedRunner$QueueUncaughtExceptionsHandler uncaughtException |  
  | WARNING: Uncaught exception in thread: Thread[#150,elasticsearch[main-cluster-1][system_write][T#1],5,TGRP-CrossClusterAsyncSearchIT] |  
  | java.lang.AssertionError: java.lang.AssertionError: listener.onFailure failed |  
  | at __randomizedtesting.SeedInfo.seed([B04664FE6047F39D]:0) |  
  | at org.elasticsearch.action.ActionListenerImplementations.safeAcceptException(ActionListenerImplementations.java:66) |  
  | at org.elasticsearch.action.ActionListener$2.onFailure(ActionListener.java:177) |  
  | at org.elasticsearch.action.bulk.TransportBulkAction.lambda$unwrappingSingleItemBulkResponse$0(TransportBulkAction.java:186) |  
  | at org.elasticsearch.action.ActionListenerImplementations$ResponseWrappingActionListener.onResponse(ActionListenerImplementations.java:173) |  
  | at org.elasticsearch.client.internal.node.NodeClient$SafelyWrappedActionListener.onResponse(NodeClient.java:160) |  
  | at org.elasticsearch.tasks.TaskManager$1.onResponse(TaskManager.java:205) |  
  | at org.elasticsearch.tasks.TaskManager$1.onResponse(TaskManager.java:199) |  
  | at org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener.onResponse(ActionListenerImplementations.java:235) |  
  | at org.elasticsearch.action.ActionListener$5.onResponse(ActionListener.java:344) |  
  | at org.elasticsearch.action.bulk.TransportBulkAction$BulkOperation$1.finishHim(TransportBulkAction.java:628) |  
  | at org.elasticsearch.action.bulk.TransportBulkAction$BulkOperation$1.onResponse(TransportBulkAction.java:609) |  
  | at org.elasticsearch.action.bulk.TransportBulkAction$BulkOperation$1.onResponse(TransportBulkAction.java:598) |  
  | at org.elasticsearch.client.internal.node.NodeClient$SafelyWrappedActionListener.onResponse(NodeClient.java:160) |  
  | at org.elasticsearch.tasks.TaskManager$1.onResponse(TaskManager.java:205) |  
  | at org.elasticsearch.tasks.TaskManager$1.onResponse(TaskManager.java:199) |  
  | at org.elasticsearch.action.support.replication.TransportReplicationAction$ReroutePhase.finishOnSuccess(TransportReplicationAction.java:1041) |  
  | at org.elasticsearch.action.support.replication.TransportReplicationAction$ReroutePhase$1.handleResponse(TransportReplicationAction.java:953) |  
  | at org.elasticsearch.action.support.replication.TransportReplicationAction$ReroutePhase$1.handleResponse(TransportReplicationAction.java:944) |  
  | at org.elasticsearch.transport.TransportService$ContextRestoreResponseHandler.handleResponse(TransportService.java:1409) |  
  | at org.elasticsearch.transport.TransportService$DirectResponseChannel.processResponse(TransportService.java:1508) | 
  | at org.elasticsearch.transport.TransportService$DirectResponseChannel.sendResponse(TransportService.java:1479) |  
  | at org.elasticsearch.transport.TaskTransportChannel.sendResponse(TaskTransportChannel.java:42) |  
  | at org.elasticsearch.action.support.ChannelActionListener.lambda$onResponse$0(ChannelActionListener.java:31) |  
  | at org.elasticsearch.action.ActionListener.run(ActionListener.java:382) |  
  | at org.elasticsearch.action.support.ChannelActionListener.onResponse(ChannelActionListener.java:31) |  
  | at org.elasticsearch.action.support.ChannelActionListener.onResponse(ChannelActionListener.java:19) |  
  | at org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener.onResponse(ActionListenerImplementations.java:235) |  
  | at org.elasticsearch.action.ActionListener$5.onResponse(ActionListener.java:344) |  
  | at org.elasticsearch.action.support.replication.TransportReplicationAction$AsyncPrimaryAction.lambda$runWithPrimaryShardReference$3(TransportReplicationAction.java:489) |  
  | at org.elasticsearch.action.ActionListener$2.onResponse(ActionListener.java:169) |  
  | at org.elasticsearch.action.ActionListenerImplementations$MappedActionListener.onResponse(ActionListenerImplementations.java:94) |  
  | at org.elasticsearch.action.support.replication.ReplicationOperation.finish(ReplicationOperation.java:474) |  
  | at org.elasticsearch.action.support.replication.ReplicationOperation.decPendingAndFinishIfNeeded(ReplicationOperation.java:460) |  
  | at org.elasticsearch.action.support.replication.ReplicationOperation$1.lambda$onResponse$0(ReplicationOperation.java:182) |  
  | at org.elasticsearch.action.support.replication.ReplicationOperation.updateCheckPoints(ReplicationOperation.java:361) | 
  | at org.elasticsearch.action.support.replication.ReplicationOperation$1.onResponse(ReplicationOperation.java:178) |  
  | at org.elasticsearch.action.support.replication.ReplicationOperation$1.onResponse(ReplicationOperation.java:173) |  
  | at org.elasticsearch.action.support.replication.TransportWriteAction$WritePrimaryResult$1.onSuccess(TransportWriteAction.java:311) |  
  | at org.elasticsearch.action.support.replication.TransportWriteAction$AsyncAfterWriteAction.maybeFinish(TransportWriteAction.java:468) |  
  | at org.elasticsearch.action.support.replication.TransportWriteAction$AsyncAfterWriteAction.run(TransportWriteAction.java:483) |  
  | at org.elasticsearch.action.support.replication.TransportWriteAction$WritePrimaryResult.runPostReplicationActions(TransportWriteAction.java:318) |  
  | at org.elasticsearch.action.support.replication.ReplicationOperation.handlePrimaryResult(ReplicationOperation.java:173) |  
  | at org.elasticsearch.action.ActionListener$2.onResponse(ActionListener.java:169) |  
  | at org.elasticsearch.action.ActionListenerImplementations$MappedActionListener.onResponse(ActionListenerImplementations.java:94) |  
  | at org.elasticsearch.action.ActionListener.completeWith(ActionListener.java:319) |  
  | at org.elasticsearch.action.bulk.TransportShardBulkAction$2.finishRequest(TransportShardBulkAction.java:273) |  
  | at org.elasticsearch.action.bulk.TransportShardBulkAction$2.doRun(TransportShardBulkAction.java:235) |  
  | at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26) |  
  | at org.elasticsearch.action.bulk.TransportShardBulkAction.performOnPrimary(TransportShardBulkAction.java:286) |  
  | at org.elasticsearch.action.bulk.TransportShardBulkAction.dispatchedShardOperationOnPrimary(TransportShardBulkAction.java:137) |  
  | at org.elasticsearch.action.bulk.TransportShardBulkAction.dispatchedShardOperationOnPrimary(TransportShardBulkAction.java:74) |  
  | at org.elasticsearch.action.support.replication.TransportWriteAction$1.doRun(TransportWriteAction.java:215) |  
  | at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26) |  
  | at org.elasticsearch.common.util.concurrent.TimedRunnable.doRun(TimedRunnable.java:33) |  
  | at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:983) |  
  | at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26) |  
  | at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) |  
  | at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) |  
  | at java.base/java.lang.Thread.run(Thread.java:1623) |  
  | Caused by: java.lang.AssertionError: listener.onFailure failed |  
  | ... 59 more |  
  | Caused by: [.async-search/3re38K96QYmul6kKEYsjCA] org.elasticsearch.index.IndexNotFoundException: no such index [.async-search] |  
  | at org.elasticsearch.indices.IndicesService.indexServiceSafe(IndicesService.java:584) |  
  | at org.elasticsearch.action.update.TransportUpdateAction.executor(TransportUpdateAction.java:96) |  
  | at org.elasticsearch.action.update.TransportUpdateAction.handleUpdateFailureWithRetry(TransportUpdateAction.java:328) |  
  | at org.elasticsearch.action.update.TransportUpdateAction.lambda$shardOperation$3(TransportUpdateAction.java:262) | 
  | at org.elasticsearch.action.ActionListenerImplementations.safeAcceptException(ActionListenerImplementations.java:60) |  
  | ... 58 more |  
  | Suppressed: [.async-search/IcYfLULQSEGBxxuw1B5zRg][[.async-search][0]] org.elasticsearch.index.engine.VersionConflictEngineException: [-AHRpoF6Q9arEqRJnDKocQ]: version conflict, required seqNo [0], primary term [1]. but no document was found
```
server/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java
response.getSeqNo(),
                                    response.getPrimaryTerm(),
                                    response.getVersion(),
                                    sourceAndContent.v2(),
                                    sourceAndContent.v1(),
                                    upsertSourceBytes
                                )
                            );
                        } else {
                            update.setGetResult(null);
                        }
                        update.setForcedRefresh(response.forcedRefresh());
                        listener.onResponse(update);
                    }, exception -> handleUpdateFailureWithRetry(listener, request, exception, retryCount)))
                );
            }
            case UPDATED -> {
                IndexRequest indexRequest = result.action();
                
                final BytesReference indexSourceBytes = indexRequest.source();
                client.bulk(
                    toSingleItemBulkRequest(indexRequest),
                    unwrappingSingleItemBulkResponse(ActionListener.<IndexResponse>wrap(response -> {
                        UpdateResponse update = new UpdateResponse(
                            response.getShardInfo(),
                            response.getShardId(),
                            response.getId(),
                            response.getSeqNo(),
                            response.getPrimaryTerm(),
                            response.getVersion(),