class ApacheHttpdLogfileInputFormat extends
        FileInputFormat<LongWritable, ParsedRecord> {

    public ApacheHttpdLogfileRecordReader createRecordReader() {
        try {
            return new ApacheHttpdLogfileRecordReader(getLogFormat(), getRequestedFields(), getTypeRemappings(), getAdditionalDissectors());
        } catch (IOException e) {
            return null;
        }
    }

}

class TestApacheHttpdLogfileInputFormat {

     String logformat;

    @Test
    public void checkAllOutputTypes() throws IOException, InterruptedException {

        Configuration conf = new Configuration(false);
        conf.set("fs.default.name", "file:///");

        // A ',' separated list of fields
        List<String> fields = Arrays.asList(
                "ANY:any",
                "ANY:any",
                "ANY:any",
                "STRING:string",
                "STRING:string",
                "STRING:string",
                "INT:int",
                "INT:int",
                "INT:int",
                "LONG:long",
                "LONG:long",
                "LONG:long",
                "FLOAT:float",
                "FLOAT:float",
                "FLOAT:float",
                "DOUBLE:double",
                "DOUBLE:double",
                "DOUBLE:double");

        File testFile = new File("src/test/resources/access.log");
        Path path = new Path(testFile.getAbsoluteFile().toURI());
        FileSplit split = new FileSplit(path, 0, testFile.length(), null);

        Map<String, Set<String>> typeRemappings = new HashMap<>();
        List<Dissector> dissectors = new ArrayList<>();
        dissectors.add(new NormalValuesDissector(HttpdLogFormatDissector.INPUT_TYPE));

        InputFormat inputFormat = new ApacheHttpdLogfileInputFormat(
            logformat,
            fields,
            typeRemappings,
            dissectors);
        TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID());
        RecordReader reader = inputFormat.createRecordReader(split, context);

        reader.initialize(split, context);

        assertTrue(reader.nextKeyValue());

        Object value = reader.getCurrentValue();
        if (value instanceof ParsedRecord) {
            ParsedRecord record = (ParsedRecord)value;
            assertEquals("42",          record.getString("ANY:any"));             // any_string
            assertEquals(42L,           record.getLong("ANY:any").longValue());   // any_long
            assertEquals(42D,           record.getDouble("ANY:any"), 0.1D);       // any_double
            assertEquals("FortyTwo",    record.getString("STRING:string"));       // string_string
            assertEquals(null,          record.getLong("STRING:string"));         // string_long
            assertEquals(null,          record.getDouble("STRING:string"));       // string_double
            assertEquals("42",          record.getString("INT:int"));             // int_string
            assertEquals(42L,           record.getLong("INT:int").longValue());   // int_long
            assertEquals(null,          record.getDouble("INT:int"));             // int_double
            assertEquals("42",          record.getString("LONG:long"));           // long_string
            assertEquals(42L,           record.getLong("LONG:long").longValue()); // long_long
            assertEquals(null,          record.getDouble("LONG:long"));           // long_double
            assertEquals("42.0",        record.getString("FLOAT:float"));         // float_string
            assertEquals(null,          record.getLong("FLOAT:float"));           // float_long
            assertEquals(42D,           record.getDouble("FLOAT:float"), 0.1D);   // float_double
            assertEquals("42.0",        record.getString("DOUBLE:double"));       // double_string
            assertEquals(null,          record.getLong("DOUBLE:double"));         // double_long
            assertEquals(42D,           record.getDouble("DOUBLE:double"), 0.1D);  // double_double
        } else {
            fail("Wrong return class type");
        }
    }

}