leave Iterables.flatten pre-caching the outer Iterable

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

@@ -54,8 +54,8 @@ public class Iterables {
         }
     }
 
-    /** Flattens the two level {@code Iterable} into a single {@code Iterable}.  Note that this uses the original input iterable so if it
-     *  later changes, the flattened result here will reflect the change. */
+    /** Flattens the two level {@code Iterable} into a single {@code Iterable}.  Note that this pre-caches the values from the outer {@code
+     *  Iterable}, but not the values from the inner one. */
     public static <T> Iterable<T> flatten(Iterable<? extends Iterable<T>> inputs) {
         Objects.requireNonNull(inputs);
         return new FlattenedIterables<>(inputs);
@@ -65,7 +65,11 @@ public class Iterables {
         private final Iterable<? extends Iterable<T>> inputs;
 
         FlattenedIterables(Iterable<? extends Iterable<T>> inputs) {
-            this.inputs = inputs;
+            List<Iterable<T>> list = new ArrayList<>();
+            for (Iterable<T> iterable : inputs) {
+                list.add(iterable);
+            }
+            this.inputs = list;
         }
 
         @Override

core/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -181,7 +181,9 @@ public class IndicesService extends AbstractLifecycleComponent<IndicesService> i
         this.namedWriteableRegistry = namedWriteableRegistry;
         clusterSettings.addSettingsUpdateConsumer(IndexStoreConfig.INDICES_STORE_THROTTLE_TYPE_SETTING, indexStoreConfig::setRateLimitingType);
         clusterSettings.addSettingsUpdateConsumer(IndexStoreConfig.INDICES_STORE_THROTTLE_MAX_BYTES_PER_SEC_SETTING, indexStoreConfig::setRateLimitingThrottle);
-        indexingMemoryController = new IndexingMemoryController(settings, threadPool, Iterables.flatten(this));
+        indexingMemoryController = new IndexingMemoryController(settings, threadPool,
+                                                                // ensure we pull an iter with new shards - flatten makes a copy
+                                                                () -> Iterables.flatten(this).iterator());
         this.indexScopeSetting = indexScopedSettings;
         this.circuitBreakerService = circuitBreakerService;
         this.indicesFieldDataCache = new IndicesFieldDataCache(settings, new IndexFieldDataCache.Listener() {

core/src/test/java/org/elasticsearch/common/util/iterable/IterablesTests.java

@@ -60,6 +60,7 @@ public class IterablesTests extends ESTestCase {
 
     public void testFlatten() {
         List<List<Integer>> list = new ArrayList<>();
+        list.add(new ArrayList<>());
 
         Iterable<Integer> allInts = Iterables.flatten(list);
         int count = 0;
@@ -67,13 +68,17 @@ public class IterablesTests extends ESTestCase {
             count++;
         }
         assertEquals(0, count);
-
         list.add(new ArrayList<>());
+        list.get(1).add(0);
+
+        // changes to the outer list are not seen since flatten pre-caches outer list on init:
+        count = 0;
         for(int x : allInts) {
             count++;
         }
         assertEquals(0, count);
 
+        // but changes to the original inner lists are seen:
         list.get(0).add(0);
         for(int x : allInts) {
             count++;