@Test public void testShuffle() throws Exception { File dir = testDir.newFolder(); File f1 = new File(dir, "file1.txt"); File f2 = new File(dir, "file2.txt"); File f3 = new File(dir, "file3.txt"); FileUtils.writeStringToFile(f1, "a,b,c", StandardCharsets.UTF_8); FileUtils.writeStringToFile(f2, "1,2,3", StandardCharsets.UTF_8); FileUtils.writeStringToFile(f3, "x,y,z", StandardCharsets.UTF_8); List<URI> uris = Arrays.asList(f1.toURI(), f2.toURI(), f3.toURI()); CSVSequenceRecordReader rr = new CSVSequenceRecordReader(); TestStreamFunction fn = new TestStreamFunction(); InputSplit is = new StreamInputSplit(uris, fn, new Random(12345)); rr.initialize(is); List<List<List<Writable>>> act = new ArrayList<>(); while (rr.hasNext()) { act.add(rr.sequenceRecord()); } rr.reset(); List<List<List<Writable>>> act2 = new ArrayList<>(); while (rr.hasNext()) { act2.add(rr.sequenceRecord()); } rr.reset(); List<List<List<Writable>>> act3 = new ArrayList<>(); while (rr.hasNext()) { act3.add(rr.sequenceRecord()); } assertEquals(3, act.size()); assertEquals(3, act2.size()); assertEquals(3, act3.size()); assertNotEquals(act, act2); assertNotEquals(act2, act3); assertNotEquals(act, act3); }
@Override public void reset() { }
StreamInputSplit implements InputSplit { @Override public void reset() { } }
StreamInputSplit implements InputSplit { @Override public void reset() { } StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn);  StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn, Random rng); }
StreamInputSplit implements InputSplit { @Override public void reset() { } StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn);  StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn, Random rng); @Override boolean canWriteToLocation(URI location); @Override String addNewLocation(); @Override String addNewLocation(String location); @Override void updateSplitLocations(boolean reset); @Override boolean needsBootstrapForWrite(); @Override void bootStrapForWrite(); @Override OutputStream openOutputStreamFor(String location); @Override InputStream openInputStreamFor(String location); @Override long length(); @Override URI[] locations(); @Override Iterator<URI> locationsIterator(); @Override Iterator<String> locationsPathIterator(); @Override void reset(); @Override boolean resetSupported(); }
StreamInputSplit implements InputSplit { @Override public void reset() { } StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn);  StreamInputSplit(@NonNull List<URI> uris, @NonNull Function<URI,InputStream> streamCreatorFn, Random rng); @Override boolean canWriteToLocation(URI location); @Override String addNewLocation(); @Override String addNewLocation(String location); @Override void updateSplitLocations(boolean reset); @Override boolean needsBootstrapForWrite(); @Override void bootStrapForWrite(); @Override OutputStream openOutputStreamFor(String location); @Override InputStream openInputStreamFor(String location); @Override long length(); @Override URI[] locations(); @Override Iterator<URI> locationsIterator(); @Override Iterator<String> locationsPathIterator(); @Override void reset(); @Override boolean resetSupported();  }