NodeIndicesStats#statsByShard is never null (#71671)

Today we support serializing `NodeIndicesStats` with a null
`statsByShard` field, but in practice it is always present (since at
least 5.6). If it were absent, we either throw NPEs or treat it as an
empty map. This commit removes the option for this field to be null on
the wire.

Spotted while researching #71670.

server/src/main/java/org/elasticsearch/indices/NodeIndicesStats.java

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.indices;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.indices.stats.CommonStats;
 import org.elasticsearch.action.admin.indices.stats.IndexShardStats;
 import org.elasticsearch.action.admin.indices.stats.ShardStats;
@@ -40,20 +41,29 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Global information on indices stats running on a specific node.
  */
 public class NodeIndicesStats implements Writeable, ToXContentFragment {
 
-    private CommonStats stats;
-    private Map<Index, List<IndexShardStats>> statsByShard;
+    private final CommonStats stats;
+    private final Map<Index, List<IndexShardStats>> statsByShard;
 
     public NodeIndicesStats(StreamInput in) throws IOException {
         stats = new CommonStats(in);
-        if (in.readBoolean()) {
+
+        final boolean hasStatsByShard;
+        if (in.getVersion().onOrAfter(Version.V_7_13_0)) {
+            hasStatsByShard = true;
+        } else {
+            hasStatsByShard = in.readBoolean();
+        }
+
+        statsByShard = new HashMap<>();
+        if (hasStatsByShard) {
             int entries = in.readVInt();
-            statsByShard = new HashMap<>();
             for (int i = 0; i < entries; i++) {
                 Index index = new Index(in);
                 int indexShardListSize = in.readVInt();
@@ -67,8 +77,7 @@ public NodeIndicesStats(StreamInput in) throws IOException {
     }
 
     public NodeIndicesStats(CommonStats oldStats, Map<Index, List<IndexShardStats>> statsByShard) {
-        //this.stats = stats;
-        this.statsByShard = statsByShard;
+        this.statsByShard = Objects.requireNonNull(statsByShard);
 
         // make a total common stats from old ones and current ones
         this.stats = oldStats;
@@ -164,15 +173,15 @@ public RecoveryStats getRecoveryStats() {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         stats.writeTo(out);
-        out.writeBoolean(statsByShard != null);
-        if (statsByShard != null) {
-            out.writeVInt(statsByShard.size());
-            for (Map.Entry<Index, List<IndexShardStats>> entry : statsByShard.entrySet()) {
-                entry.getKey().writeTo(out);
-                out.writeVInt(entry.getValue().size());
-                for (IndexShardStats indexShardStats : entry.getValue()) {
-                    indexShardStats.writeTo(out);
-                }
+        if (out.getVersion().before(Version.V_7_13_0)) {
+            out.writeBoolean(true);
+        }
+        out.writeVInt(statsByShard.size());
+        for (Map.Entry<Index, List<IndexShardStats>> entry : statsByShard.entrySet()) {
+            entry.getKey().writeTo(out);
+            out.writeVInt(entry.getValue().size());
+            for (IndexShardStats indexShardStats : entry.getValue()) {
+                indexShardStats.writeTo(out);
             }
         }
     }
@@ -237,11 +246,7 @@ private Map<Index, CommonStats> createStatsByIndex() {
     }
 
     public List<IndexShardStats> getShardStats(Index index) {
-        if (statsByShard == null) {
-            return null;
-        } else {
-            return statsByShard.get(index);
-        }
+        return statsByShard.get(index);
     }
 
     static final class Fields {