Iterables.flatten should not pre-cache the first iterator

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

@@ -63,11 +63,7 @@ public class Iterables {
         private final Iterable<? extends Iterable<T>> inputs;
 
         FlattenedIterables(Iterable<? extends Iterable<T>> inputs) {
-            List<Iterable<T>> list = new ArrayList<>();
-            for (Iterable<T> iterable : inputs) {
-                list.add(iterable);
-            }
-            this.inputs = list;
+            this.inputs = inputs;
         }
 
         @Override

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

@@ -93,7 +93,7 @@ public class IndexingMemoryController extends AbstractComponent implements Index
 
     private final ShardsIndicesStatusChecker statusChecker;
 
-    IndexingMemoryController(Settings settings, ThreadPool threadPool, Iterable<IndexShard>indexServices) {
+    IndexingMemoryController(Settings settings, ThreadPool threadPool, Iterable<IndexShard> indexServices) {
         this(settings, threadPool, indexServices, JvmInfo.jvmInfo().getMem().getHeapMax().bytes());
     }
 

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

@@ -19,12 +19,14 @@
 
 package org.elasticsearch.common.util.iterable;
 
-import org.elasticsearch.test.ESTestCase;
-
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
+import java.util.List;
 import java.util.NoSuchElementException;
 
+import org.elasticsearch.test.ESTestCase;
+
 import static org.hamcrest.object.HasToString.hasToString;
 
 public class IterablesTests extends ESTestCase {
@@ -56,6 +58,29 @@ public class IterablesTests extends ESTestCase {
         test(iterable);
     }
 
+    public void testFlatten() {
+        List<List<Integer>> list = new ArrayList<>();
+
+        Iterable<Integer> allInts = Iterables.flatten(list);
+        int count = 0;
+        for(int x : allInts) {
+            count++;
+        }
+        assertEquals(0, count);
+
+        list.add(new ArrayList<>());
+        for(int x : allInts) {
+            count++;
+        }
+        assertEquals(0, count);
+
+        list.get(0).add(0);
+        for(int x : allInts) {
+            count++;
+        }
+        assertEquals(1, count);
+    }
+
     private void test(Iterable<String> iterable) {
         try {
             Iterables.get(iterable, -1);
@@ -73,4 +98,4 @@ public class IterablesTests extends ESTestCase {
             assertThat(e, hasToString("java.lang.IndexOutOfBoundsException: 3"));
         }
     }
-}
+}