Iterable instead

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/routing/RoutingNode.java

@@ -10,8 +10,8 @@
 package org.elasticsearch.cluster.routing;
 
 import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.common.util.iterable.Iterables;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.shard.ShardId;
@@ -214,16 +214,16 @@ void remove(ShardRouting shard) {
         assert invariant();
     }
 
-    public Iterator<ShardRouting> initializing() {
-        return Iterators.assertReadOnly(initializingShards.iterator());
+    public Iterable<ShardRouting> initializing() {
+        return Iterables.assertReadOnly(initializingShards);
     }
 
-    public Iterator<ShardRouting> relocating() {
-        return Iterators.assertReadOnly(relocatingShards.iterator());
+    public Iterable<ShardRouting> relocating() {
+        return Iterables.assertReadOnly(relocatingShards);
     }
 
-    public Iterator<ShardRouting> started() {
-        return Iterators.assertReadOnly(startedShards.iterator());
+    public Iterable<ShardRouting> started() {
+        return Iterables.assertReadOnly(startedShards);
     }
 
     /**

server/src/main/java/org/elasticsearch/cluster/routing/allocation/RoutingAllocation.java

@@ -175,9 +175,7 @@ private static Map<String, Long> unaccountedSearchableSnapshotSizes(ClusterState
                 DiskUsage usage = clusterInfo.getNodeMostAvailableDiskUsages().get(node.nodeId());
                 ClusterInfo.ReservedSpace reservedSpace = clusterInfo.getReservedSpace(node.nodeId(), usage != null ? usage.path() : "");
                 long totalSize = 0;
-                final var startedShardsIterator = node.started();
-                while (startedShardsIterator.hasNext()) {
-                    final var shard = startedShardsIterator.next();
+                for (ShardRouting shard : node.started()) {
                     if (shard.getExpectedShardSize() > 0
                         && clusterState.metadata().indexMetadata(shard.index()).isSearchableSnapshot()
                         && reservedSpace.containsShardId(shard.shardId()) == false

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/DesiredBalanceComputer.java

@@ -298,7 +298,7 @@ public DesiredBalance compute(
                     if (rerouteExplanation.decisions().type() != Decision.Type.NO) {
                         final Iterator<ShardRouting> initializingShardsIterator = routingNodes.node(
                             routingAllocation.nodes().resolveNode(command.toNode()).getId()
-                        ).initializing();
+                        ).initializing().iterator();
                         assert initializingShardsIterator.hasNext();
                         final var initializingShard = initializingShardsIterator.next();
                         assert initializingShardsIterator.hasNext() == false
@@ -538,11 +538,11 @@ static void maybeSimulateAlreadyStartedShards(
         // Find all shards that are started in RoutingNodes but have no data on corresponding node in ClusterInfo
         final var startedShards = new ArrayList<ShardRouting>();
         for (var routingNode : routingNodes) {
-            routingNode.started().forEachRemaining(shardRouting -> {
+            for (var shardRouting : routingNode.started()) {
                 if (clusterInfo.hasShardMoved(shardRouting)) {
                     startedShards.add(shardRouting);
                 }
-            });
+            }
         }
         if (startedShards.isEmpty()) {
             return;

server/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDecider.java

@@ -108,9 +108,7 @@ public static long sizeOfUnaccountedShards(
         // no longer initializing because their recovery failed or was cancelled.
 
         // Where reserved space is unavailable (e.g. stats are out-of-sync) compute a conservative estimate for initialising shards
-        final var initializingShardsIterator = node.initializing();
-        while (initializingShardsIterator.hasNext()) {
-            final var routing = initializingShardsIterator.next();
+        for (ShardRouting routing : node.initializing()) {
             // Space needs to be reserved only when initializing shards that are going to use additional space
             // that is not yet accounted for by `reservedSpace` in case of lengthy recoveries
             if (shouldReserveSpaceForInitializingShard(routing, metadata) && reservedSpace.containsShardId(routing.shardId()) == false) {
@@ -137,9 +135,7 @@ public static long sizeOfUnaccountedShards(
         totalSize += sizeOfUnaccountableSearchableSnapshotShards;
 
         if (subtractShardsMovingAway) {
-            final var relocatingShardsIterator = node.relocating();
-            while (relocatingShardsIterator.hasNext()) {
-                final var routing = relocatingShardsIterator.next();
+            for (ShardRouting routing : node.relocating()) {
                 if (dataPath.equals(clusterInfo.getDataPath(routing))) {
                     ProjectMetadata project = metadata.projectFor(routing.index());
                     totalSize -= getExpectedShardSize(

server/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/ThrottlingAllocationDecider.java

@@ -127,10 +127,8 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
 
             // Count the primaries currently doing recovery on the node, to ensure the primariesInitialRecoveries setting is obeyed.
             int primariesInRecovery = 0;
-            final var initializingShardsIterator = node.initializing();
             final var returnUnexplainedDecision = allocation.debugDecision() == false;
-            while (initializingShardsIterator.hasNext()) {
-                final var shard = initializingShardsIterator.next();
+            for (ShardRouting shard : node.initializing()) {
                 // when a primary shard is INITIALIZING, it can be because of *initial recovery* or *relocation from another node*
                 // we only count initial recoveries here, so we need to make sure that relocating node is null
                 if (shard.primary() && shard.relocatingNodeId() == null) {

server/src/main/java/org/elasticsearch/common/util/iterable/Iterables.java

@@ -9,6 +9,9 @@
 
 package org.elasticsearch.common.util.iterable;
 
+import org.elasticsearch.common.collect.Iterators;
+import org.elasticsearch.core.Assertions;
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -81,4 +84,15 @@ public static <T> int indexOf(Iterable<T> iterable, Predicate<T> predicate) {
     public static long size(Iterable<?> iterable) {
         return StreamSupport.stream(iterable.spliterator(), false).count();
     }
+
+    /**
+     * Adds a wrapper around {@code iterable} which asserts that {@link Iterator#remove()} is not called on the iterator it returns.
+     */
+    public static <T> Iterable<T> assertReadOnly(Iterable<T> iterable) {
+        if (Assertions.ENABLED) {
+            return () -> Iterators.assertReadOnly(iterable.iterator());
+        } else {
+            return iterable;
+        }
+    }
 }

server/src/test/java/org/elasticsearch/cluster/routing/RoutingNodeTests.java

@@ -161,7 +161,7 @@ public void testReturnStartedShards() {
 
     private static Set<ShardId> startedShardsSet(RoutingNode routingNode) {
         final var result = new HashSet<ShardId>();
-        routingNode.started().forEachRemaining(shardRouting -> result.add(shardRouting.shardId()));
+        routingNode.started().forEach(shardRouting -> result.add(shardRouting.shardId()));
         return result;
     }
 

server/src/test/java/org/elasticsearch/cluster/routing/allocation/allocator/DesiredBalanceComputerTests.java

@@ -582,7 +582,7 @@ public void allocate(RoutingAllocation allocation) {
                     allocation.routingNodes().getRelocatingShardCount(),
                     equalTo(0)
                 );
-                assertThat(Iterators.toList(allocation.routingNodes().node("node-2").started()), hasSize(2));
+                assertThat(Iterators.toList(allocation.routingNodes().node("node-2").started().iterator()), hasSize(2));
             }
 
             @Override