### Question:
RecordReaderFunction implements Function<Tuple2<String, PortableDataStream>, List<Writable>> { public RecordReaderFunction(RecordReader recordReader) { this.recordReader = recordReader; } RecordReaderFunction(RecordReader recordReader); @Override List<Writable> call(Tuple2<String, PortableDataStream> value);  }### Answer:
@Test public void testRecordReaderFunction() throws Exception { File f = testDir.newFolder(); new ClassPathResource("datavec-spark/imagetest/").copyDirectory(f); List<String> labelsList = Arrays.asList("0", "1"); String path = f.getAbsolutePath() + "/*"; JavaPairRDD<String, PortableDataStream> origData = sc.binaryFiles(path); assertEquals(4, origData.count()); ImageRecordReader irr = new ImageRecordReader(28, 28, 1, new ParentPathLabelGenerator()); irr.setLabels(labelsList); RecordReaderFunction rrf = new RecordReaderFunction(irr); JavaRDD<List<Writable>> rdd = origData.map(rrf); List<List<Writable>> listSpark = rdd.collect(); assertEquals(4, listSpark.size()); for (int i = 0; i < 4; i++) { assertEquals(1 + 1, listSpark.get(i).size()); assertEquals(28 * 28, ((ArrayWritable) listSpark.get(i).iterator().next()).length()); } InputSplit is = new FileSplit(f, new String[] {"bmp"}, true); irr = new ImageRecordReader(28, 28, 1, new ParentPathLabelGenerator()); irr.initialize(is); List<List<Writable>> list = new ArrayList<>(4); while (irr.hasNext()) { list.add(irr.next()); } assertEquals(4, list.size()); boolean[] found = new boolean[4]; for (int i = 0; i < 4; i++) { int foundIndex = -1; List<Writable> collection = listSpark.get(i); for (int j = 0; j < 4; j++) { if (collection.equals(list.get(j))) { if (foundIndex != -1) fail(); foundIndex = j; if (found[foundIndex]) fail(); found[foundIndex] = true; } } } int count = 0; for (boolean b : found) if (b) count++; assertEquals(4, count); }