private void validateFileSplits(final List<FileSplit> fileSplits, final Configuration configuration,
                                           final Class<? extends InputFormat<NullWritable, VertexWritable>> inputFormatClass,
                                           final Optional<Class<? extends OutputFormat<NullWritable, VertexWritable>>> outFormatClass) throws Exception {

        final InputFormat inputFormat = ReflectionUtils.newInstance(inputFormatClass, configuration);
        final TaskAttemptContext job = new TaskAttemptContextImpl(configuration, new TaskAttemptID(UUID.randomUUID().toString(), 0, TaskType.MAP, 0, 0));

        int vertexCount = 0;
        int outEdgeCount = 0;
        int inEdgeCount = 0;

        final OutputFormat<NullWritable, VertexWritable> outputFormat = outFormatClass.isPresent() ? ReflectionUtils.newInstance(outFormatClass.get(), configuration) : null;
        final RecordWriter<NullWritable, VertexWritable> writer = null == outputFormat ? null : outputFormat.getRecordWriter(job);

        GraphFilter graphFilter = null;
        if (configuration.get(Constants.GREMLIN_HADOOP_GRAPH_FILTER, null) != null) {
            graphFilter = VertexProgramHelper.deserialize(ConfUtil.makeApacheConfiguration(configuration),
                    Constants.GREMLIN_HADOOP_GRAPH_FILTER);
        }

        boolean foundKeyValue = false;
        for (final FileSplit split : fileSplits) {
            logger.info("\treading file split {}", split.getPath().getName() + " ({}", split.getStart() + "..." + (split.getStart() + split.getLength()), "{} {} bytes)");
            final RecordReader reader = inputFormat.createRecordReader(split, job);

            float lastProgress = -1f;
            while (reader.nextKeyValue()) {
                //System.out.println("" + reader.getProgress() + "> " + reader.getCurrentKey() + ": " + reader.getCurrentValue());
                final float progress = reader.getProgress();
                assertTrue(progress >= lastProgress);
                assertEquals(NullWritable.class, reader.getCurrentKey().getClass());
                final VertexWritable vertexWritable = (VertexWritable) reader.getCurrentValue();
                if (graphFilter != null) {
                    // check whether the filter takes effect
                    assertTrue(vertexWritable.get().applyGraphFilter(graphFilter).isPresent());
                }
                if (null != writer) writer.write(NullWritable.get(), vertexWritable);
                vertexCount++;
                outEdgeCount = outEdgeCount + (int) IteratorUtils.count(vertexWritable.get().edges(Direction.OUT));
                inEdgeCount = inEdgeCount + (int) IteratorUtils.count(vertexWritable.get().edges(Direction.IN));
                //
                final Vertex vertex = vertexWritable.get();
                if (getInputFilename().startsWith("grateful-dead-v")) {
                    assertEquals(Integer.class, vertex.id().getClass());
                    if (vertex.value("name").equals("SUGAR MAGNOLIA")) {
                        foundKeyValue = true;
                        assertEquals(92, IteratorUtils.count(vertex.edges(Direction.OUT)));
                        assertEquals(77, IteratorUtils.count(vertex.edges(Direction.IN)));
                    }
                } else if (getInputFilename().startsWith("tinkerpop-classic-byteid-")) {
                    assertTrue(vertex.id().getClass().equals(Byte.class));
                }
                lastProgress = progress;
            }
        }

        if (getInputFilename().startsWith("grateful-dead-v") && graphFilter == null) {
            assertEquals(8049, outEdgeCount);
            assertEquals(8049, inEdgeCount);
            assertEquals(808, vertexCount);
            assertTrue(foundKeyValue);
        }
        if (graphFilter == null) {
            assertEquals(outEdgeCount, inEdgeCount);
        }

        if (null != writer) {
            writer.close(new TaskAttemptContextImpl(configuration, job.getTaskAttemptID()));
            for (int i = 1; i < 10; i++) {
                final File outputDirectory = new File(new URL(configuration.get("mapreduce.output.fileoutputformat.outputdir")).toURI());
                final List<FileSplit> splits = generateFileSplits(new File(outputDirectory.getAbsoluteFile() + "/_temporary/0/_temporary/" + job.getTaskAttemptID().getTaskID().toString().replace("task", "attempt") + "_0" + "/part-m-00000"), i);
                validateFileSplits(splits, configuration, inputFormatClass, Optional.empty());
            }
        }
    }
@author Daniel Kuppitz (http://gremlin.guru)
@author Marko A. Rodriguez (http://markorodriguez.com)
hadoop-gremlin/src/test/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/RecordReaderWriterTest.java