Repository: deeplearning4j/deeplearning4j
Issue: InputStream used twice, causing a NoSuchElementException

In case data is contained in an InputStream, the following snippet fails:
```
InputStream dataFile = this.getClass().getResourceAsStream("/mydata.csv");
RecordReader recordReader = new CSVRecordReader(0, ",");
recordReader.initialize(new InputStreamInputSplit(dataFile));
DataSetIterator iterator = new RecordReaderDataSetIterator(recordReader, batchSize, labelIndex, numClasses);
DataSet allData = iterator.next();
```

The exception is the following:
```
Exception in thread "Thread-3" java.util.NoSuchElementException: No next elements
        at org.deeplearning4j.datasets.datavec.RecordReaderMultiDataSetIterator.next(RecordReaderMultiDataSetIterator.java:116)
        at org.deeplearning4j.datasets.datavec.RecordReaderDataSetIterator.next(RecordReaderDataSetIterator.java:320)
        at org.deeplearning4j.datasets.datavec.RecordReaderDataSetIterator.next(RecordReaderDataSetIterator.java:409)
        at org.deeplearning4j.datasets.datavec.RecordReaderDataSetIterator.next(RecordReaderDataSetIterator.java:51)
        at com.gluonhq.dlearning.DataRetrievalService.getData(DataRetrievalService.java:92)

```

Looking into the code, it seems the `InputStream` is consumed twice, by 2 different iterators. As a result, the second iterator will return false in its `hasNext()` method.

The first iterator is created in 
`recordReader.initialize(new InputStreamInputSplit(dataFile));
`
The initialize method will set the InputStreamInputSplit, but also create the iterator. I don't see a method on recordReader that sets the InputStreamInputSplit but doesn't initialize the iterator.

Next, the second iterator is created around the same InputStream in `org.datavec.api.records.reader.impl.LineRecordReader` 
when the next method is called:
`
iter = IOUtils.lineIterator(new InputStreamReader(locations[splitIndex].toURL().openStream()));`



Changed files:
deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderMultiDataSetIterator.java
deeplearning4j-scaleout/spark/dl4j-spark/src/main/java/org/deeplearning4j/spark/datavec/iterator/SparkSourceDummyReader.java
deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetIterator.java
deeplearning4j-core/src/test/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetiteratorTest.java

diff --git a/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetIterator.java b/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetIterator.java
index f480864109..a382b84253 100644
--- a/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetIterator.java
+++ b/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetIterator.java
@@ -27,6 +27,8 @@ import org.datavec.api.records.metadata.RecordMetaData;
 import org.datavec.api.records.metadata.RecordMetaDataComposableMap;
 import org.datavec.api.records.reader.RecordReader;
 import org.datavec.api.records.reader.SequenceRecordReader;
+import org.datavec.api.records.reader.impl.ConcatenatingRecordReader;
+import org.datavec.api.records.reader.impl.collection.CollectionRecordReader;
 import org.datavec.api.writable.Writable;
 import org.nd4j.linalg.api.ndarray.INDArray;
 import org.nd4j.linalg.dataset.DataSet;
@@ -170,6 +172,13 @@ public class RecordReaderDataSetIterator implements DataSetIterator {
         this.collectMetaData = collectMetaData;
     }
 
+    private void initializeUnderlying(){
+        if (underlying == null) {
+            Record next = recordReader.nextRecord();
+            initializeUnderlying(next);
+        }
+    }
+
     private void initializeUnderlying(Record next) {
         int totalSize = next.getRecord().size();
 
@@ -178,7 +187,15 @@ public class RecordReaderDataSetIterator implements DataSetIterator {
             labelIndex = totalSize - 1;
         }
 
-        recordReader.reset();
+        if(recordReader.resetSupported()) {
+            recordReader.reset();
+        } else {
+            //Hack around the fact that we need the first record to initialize the underlying RRMDSI, but can't reset
+            // the original reader
+            recordReader = new ConcatenatingRecordReader(
+                    new CollectionRecordReader(Collections.singletonList(next.getRecord())),
+                    recordReader);
+        }
 
         RecordReaderMultiDataSetIterator.Builder builder = new RecordReaderMultiDataSetIterator.Builder(batchSize);
         if (recordReader instanceof SequenceRecordReader) {
@@ -297,8 +314,7 @@ public class RecordReaderDataSetIterator implements DataSetIterator {
         }
 
         if (underlying == null) {
-            Record next = recordReader.nextRecord();
-            initializeUnderlying(next);
+            initializeUnderlying();
         }
 
 
@@ -343,7 +359,10 @@ public class RecordReaderDataSetIterator implements DataSetIterator {
 
     @Override
     public boolean resetSupported() {
-        return true;
+        if(underlying == null){
+            initializeUnderlying();
+        }
+        return underlying.resetSupported();
     }
 
     @Override
diff --git a/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderMultiDataSetIterator.java b/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderMultiDataSetIterator.java
index 527a4e192a..ebb31158e8 100644
--- a/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderMultiDataSetIterator.java
+++ b/deeplearning4j-core/src/main/java/org/deeplearning4j/datasets/datavec/RecordReaderMultiDataSetIterator.java
@@ -86,6 +86,8 @@ public class RecordReaderMultiDataSetIterator implements MultiDataSetIterator, S
 
     private MultiDataSetPreProcessor preProcessor;
 
+    private boolean resetSupported = true;
+
     private RecordReaderMultiDataSetIterator(Builder builder) {
         this.batchSize = builder.batchSize;
         this.alignmentMode = builder.alignmentMode;
@@ -97,6 +99,18 @@ public class RecordReaderMultiDataSetIterator implements MultiDataSetIterator, S
         if (this.timeSeriesRandomOffset) {
             timeSeriesRandomOffsetRng = new Random(builder.timeSeriesRandomOffsetSeed);
         }
+
+
+        if(recordReaders != null){
+            for(RecordReader rr : recordReaders.values()){
+                resetSupported &= rr.resetSupported();
+            }
+        }
+        if(sequenceRecordReaders != null){
+            for(SequenceRecordReader srr : sequenceRecordReaders.values()){
+                resetSupported &= srr.resetSupported();
+            }
+        }
     }
 
     @Override
@@ -622,7 +636,7 @@ public class RecordReaderMultiDataSetIterator implements MultiDataSetIterator, S
 
     @Override
     public boolean resetSupported() {
-        return true;
+        return resetSupported;
     }
 
     @Override
@@ -632,6 +646,11 @@ public class RecordReaderMultiDataSetIterator implements MultiDataSetIterator, S
 
     @Override
     public void reset() {
+        if(!resetSupported){
+            throw new IllegalStateException("Cannot reset iterator - reset not supported (resetSupported() == false):" +
+                    " one or more underlying (sequence) record readers do not support resetting");
+        }
+
         for (RecordReader rr : recordReaders.values())
             rr.reset();
         for (SequenceRecordReader rr : sequenceRecordReaders.values())
diff --git a/deeplearning4j-core/src/test/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetiteratorTest.java b/deeplearning4j-core/src/test/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetiteratorTest.java
index 857989f5ae..c4cc460577 100644
--- a/deeplearning4j-core/src/test/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetiteratorTest.java
+++ b/deeplearning4j-core/src/test/java/org/deeplearning4j/datasets/datavec/RecordReaderDataSetiteratorTest.java
@@ -29,6 +29,7 @@ import org.datavec.api.records.reader.impl.collection.CollectionSequenceRecordRe
 import org.datavec.api.records.reader.impl.csv.CSVRecordReader;
 import org.datavec.api.records.reader.impl.csv.CSVSequenceRecordReader;
 import org.datavec.api.split.FileSplit;
+import org.datavec.api.split.InputStreamInputSplit;
 import org.datavec.api.split.NumberedFileInputSplit;
 import org.datavec.api.writable.DoubleWritable;
 import org.datavec.api.writable.IntWritable;
@@ -1248,4 +1249,35 @@ public class RecordReaderDataSetiteratorTest {
         iter.reset();
         iter.next();
     }
+
+    @Test
+    public void testReadingFromStream() throws Exception {
+        int batchSize = 1;
+        int labelIndex = 4;
+        int numClasses = 3;
+        InputStream dataFile = new ClassPathResource("iris.txt").getInputStream();
+        RecordReader recordReader = new CSVRecordReader(0, ',');
+        recordReader.initialize(new InputStreamInputSplit(dataFile));
+
+        assertTrue(recordReader.hasNext());
+        assertFalse(recordReader.resetSupported());
+
+        DataSetIterator iterator = new RecordReaderDataSetIterator(recordReader, batchSize, labelIndex, numClasses);
+        assertFalse(iterator.resetSupported());
+
+        int count = 0;
+        while(iterator.hasNext()){
+            assertNotNull(iterator.next());
+            count++;
+        }
+
+        assertEquals(150, count);
+
+        try{
+            iterator.reset();
+            fail("Expected exception");
+        } catch (Exception e){
+            //expected
+        }
+    }
 }
diff --git a/deeplearning4j-scaleout/spark/dl4j-spark/src/main/java/org/deeplearning4j/spark/datavec/iterator/SparkSourceDummyReader.java b/deeplearning4j-scaleout/spark/dl4j-spark/src/main/java/org/deeplearning4j/spark/datavec/iterator/SparkSourceDummyReader.java
index 27d866ead8..c1b81a162d 100644
--- a/deeplearning4j-scaleout/spark/dl4j-spark/src/main/java/org/deeplearning4j/spark/datavec/iterator/SparkSourceDummyReader.java
+++ b/deeplearning4j-scaleout/spark/dl4j-spark/src/main/java/org/deeplearning4j/spark/datavec/iterator/SparkSourceDummyReader.java
@@ -75,6 +75,11 @@ public class SparkSourceDummyReader implements RecordReader, Serializable {
     @Override
     public void reset() { /* No op */ }
 
+    @Override
+    public boolean resetSupported() {
+        return true;
+    }
+
     @Override
     public List<Writable> record(URI uri, DataInputStream dataInputStream) throws IOException {
         throw new UnsupportedOperationException();