### Elasticsearch Version

8.2.0 (likely others)

### Installed Plugins

None

### Java Version

_bundled_

### OS Version

Cloud

### Problem Description

I saw an NPE in `DataTiersUsageTransportAction` which I believe was because the shards whose stats were collected in the nodes stats differed from those in the cluster state, which could happen due to a concurrent shard movement.

### Steps to Reproduce

Not known

### Logs (if relevant)

```
collector [cluster_stats] failed to collect data
java.lang.NullPointerException: Cannot invoke "org.elasticsearch.cluster.routing.ShardRouting.state()" because the return value of "org.elasticsearch.cluster.routing.RoutingNode.getByShardId(org.elasticsearch.index.shard.ShardId)" is null
	at org.elasticsearch.xpack.core.DataTiersUsageTransportAction.aggregateDataTierShardStats(DataTiersUsageTransportAction.java:210) ~[?:?]
	at org.elasticsearch.xpack.core.DataTiersUsageTransportAction.classifyIndexAndCollectStats(DataTiersUsageTransportAction.java:194) ~[?:?]
	at org.elasticsearch.xpack.core.DataTiersUsageTransportAction.lambda$aggregateDataTierIndexStats$4(DataTiersUsageTransportAction.java:175) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[?:?]
	at java.util.stream.DistinctOps$1$2.accept(DistinctOps.java:174) ~[?:?]
	at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.util.Iterator.forEachRemaining(Iterator.java:133) ~[?:?]
	at java.util.Collections$UnmodifiableCollection$1.forEachRemaining(Collections.java:1061) ~[?:?]
	at java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1845) ~[?:?]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) ~[?:?]
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) ~[?:?]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) ~[?:?]
	at org.elasticsearch.xpack.core.DataTiersUsageTransportAction.aggregateDataTierIndexStats(DataTiersUsageTransportAction.java:175) ~[?:?]
	at org.elasticsearch.xpack.core.DataTiersUsageTransportAction.calculateStats(DataTiersUsageTransportAction.java:140) ~[?:?]
	at org.elasticsearch.xpack.core.DataTiersUsageTransportAction.lambda$masterOperation$0(DataTiersUsageTransportAction.java:91) ~[?:?]
	at org.elasticsearch.action.ActionListener$2.onResponse(ActionListener.java:162) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.client.internal.node.NodeClient$ActionResponseTaskListener.onResponse(NodeClient.java:175) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.tasks.TaskManager$1.onResponse(TaskManager.java:176) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.tasks.TaskManager$1.onResponse(TaskManager.java:170) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.action.support.ContextPreservingActionListener.onResponse(ContextPreservingActionListener.java:31) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.lambda$applyInternal$2(SecurityActionFilter.java:163) ~[?:?]
	at org.elasticsearch.action.ActionListener$DelegatingFailureActionListener.onResponse(ActionListener.java:245) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.action.ActionListener.completeWith(ActionListener.java:473) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.action.support.nodes.TransportNodesAction.newResponseAsync(TransportNodesAction.java:181) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.action.support.nodes.TransportNodesAction.newResponse(TransportNodesAction.java:156) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.action.support.nodes.TransportNodesAction$AsyncAction.lambda$finishHim$0(TransportNodesAction.java:303) ~[elasticsearch-8.2.0.jar:8.2.0]
	at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:714) [elasticsearch-8.2.0.jar:8.2.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
	at java.lang.Thread.run(Thread.java:833) [?:?]
```
x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/DataTiersUsageTransportAction.java
static Map<String, DataTiersFeatureSetUsage.TierSpecificStats> calculateStats(
        List<NodeStats> nodesStats,
        Map<String, String> indexByTier,
        RoutingNodes routingNodes
    ) {
        Map<String, TierStatsAccumulator> statsAccumulators = new HashMap<>();
        for (NodeStats nodeStats : nodesStats) {
            aggregateDataTierNodeCounts(nodeStats, statsAccumulators);
            aggregateDataTierIndexStats(nodeStats, routingNodes, indexByTier, statsAccumulators);
        }
        Map<String, DataTiersFeatureSetUsage.TierSpecificStats> results = new HashMap<>();
        for (Map.Entry<String, TierStatsAccumulator> entry : statsAccumulators.entrySet()) {
            results.put(entry.getKey(), calculateFinalTierStats(entry.getValue()));
        }
        return results;
    }
    /**
     * Determine which data tiers this node belongs to (if any), and increment the node counts for those tiers.
     */
    private static void aggregateDataTierNodeCounts(NodeStats nodeStats, Map<String, TierStatsAccumulator> tiersStats) {
        nodeStats.getNode()
            .getRoles()
            .stream()
            .map(DiscoveryNodeRole::roleName)
            .filter(DataTier::validTierName)
            .forEach(tier -> tiersStats.computeIfAbsent(tier, k -> new TierStatsAccumulator()).nodeCount++);
    }
    /**
     * Locate which indices are hosted on the node specified by the NodeStats, then group and aggregate the available index stats by tier.