**Build scan:**
https://gradle-enterprise.elastic.co/s/bcxstpjolnb2q/tests/:x-pack:plugin:ccr:internalClusterTest/org.elasticsearch.xpack.ccr.FollowerFailOverIT/testReadRequestsReturnLatestMappingVersion

**Reproduction line:**
```
./gradlew ':x-pack:plugin:ccr:internalClusterTest' --tests "org.elasticsearch.xpack.ccr.FollowerFailOverIT.testReadRequestsReturnLatestMappingVersion" -Dtests.seed=A75C1F0C6F541957 -Dtests.locale=fr-CA -Dtests.timezone=America/Tijuana -Druntime.java=20
```

**Applicable branches:**
8.10

**Reproduces locally?:**
Didn't try

**Failure history:**
https://gradle-enterprise.elastic.co/scans/tests?tests.container=org.elasticsearch.xpack.ccr.FollowerFailOverIT&tests.test=testReadRequestsReturnLatestMappingVersion

**Failure excerpt:**
```
com.carrotsearch.randomizedtesting.UncaughtExceptionError: Captured an uncaught exception in thread: Thread[id=498, name=elasticsearch[follower2][clusterApplierService#updateTask][T#1], state=RUNNABLE, group=TGRP-FollowerFailOverIT]

  at __randomizedtesting.SeedInfo.seed([A75C1F0C6F541957:A564CB31A58082CC]:0)

  Caused by: java.lang.AssertionError: (No message provided)

    at __randomizedtesting.SeedInfo.seed([A75C1F0C6F541957]:0)
    at org.elasticsearch.discovery.PeerFinder.closePeers(PeerFinder.java:182)
    at org.elasticsearch.cluster.coordination.Coordinator.onClusterStateApplied(Coordinator.java:415)
    at org.elasticsearch.cluster.coordination.Coordinator.lambda$handleApplyCommit$8(Coordinator.java:403)
    at org.elasticsearch.action.ActionListenerImplementations$MappedActionListener.lambda$map$0(ActionListenerImplementations.java:108)
    at org.elasticsearch.action.ActionListenerImplementations$MappedActionListener.onResponse(ActionListenerImplementations.java:89)
    at org.elasticsearch.cluster.service.ClusterApplierService$ClusterApplyActionListener.onResponse(ClusterApplierService.java:598)
    at org.elasticsearch.cluster.service.ClusterApplierService$ClusterApplyActionListener.onResponse(ClusterApplierService.java:569)
    at org.elasticsearch.cluster.service.ClusterApplierService.runTask(ClusterApplierService.java:439)
    at org.elasticsearch.cluster.service.ClusterApplierService$UpdateTask.run(ClusterApplierService.java:154)
    at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:916)
    at org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.runAndClean(PrioritizedEsThreadPoolExecutor.java:217)
    at org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.run(PrioritizedEsThreadPoolExecutor.java:183)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
    at java.lang.Thread.run(Thread.java:1623)

```
server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java
final Optional<DiscoveryNode> peerFinderLeader = peerFinder.getLeader();
            final ClusterState lastAcceptedClusterState = getStateForMasterService();
            assert peerFinder.getCurrentTerm() == getCurrentTerm();
            assert followersChecker.getFastResponseState().term() == getCurrentTerm() : followersChecker.getFastResponseState();
            assert followersChecker.getFastResponseState().mode() == getMode() : followersChecker.getFastResponseState();
            assert (applierState.nodes().getMasterNodeId() == null) == applierState.blocks().hasGlobalBlockWithId(NO_MASTER_BLOCK_ID);
            assert preVoteCollector.getPreVoteResponse().equals(getPreVoteResponse()) : preVoteCollector + " vs " + getPreVoteResponse();
            assert lagDetector.getTrackedNodes().contains(getLocalNode()) == false : lagDetector.getTrackedNodes();
            assert followersChecker.getKnownFollowers().equals(lagDetector.getTrackedNodes())
                : followersChecker.getKnownFollowers() + " vs " + lagDetector.getTrackedNodes();
            assert singleNodeClusterChecker == null || (mode == Mode.LEADER && applierState.nodes().size() == 1)
                : "Single node checker must exist iff there is a single-node cluster";
            if (mode == Mode.LEADER) {
                final boolean becomingMaster = lastAcceptedClusterState.term() != getCurrentTerm();
                assert coordinationState.get().electionWon();
                assert lastKnownLeader.isPresent() && lastKnownLeader.get().equals(getLocalNode());
                assert joinAccumulator instanceof JoinHelper.LeaderJoinAccumulator;
                assert peerFinderLeader.equals(lastKnownLeader) : peerFinderLeader;
                assert prevotingRound == null : prevotingRound;
                assert becomingMaster || lastAcceptedClusterState.nodes().getMasterNodeId() != null : lastAcceptedClusterState;
                assert leaderChecker.leader() == null : leaderChecker.leader();
                assert getLocalNode().equals(applierState.nodes().getMasterNode())
                    || (applierState.nodes().getMasterNodeId() == null && applierState.term() < getCurrentTerm());
                assert preVoteCollector.getLeader() == getLocalNode() : preVoteCollector;
                assert clusterFormationFailureHelper.isRunning() == false;
                final boolean activePublication = currentPublication.map(CoordinatorPublication::isActiveForCurrentLeader).orElse(false);
                if (becomingMaster && activePublication == false) {
                    
                    assert followersChecker.getKnownFollowers().isEmpty() : followersChecker.getKnownFollowers();
                } else {