Use generic APIs for more type safety

[fmbenhassine]

The Record<P> API has been generic from its inception, since the record's payload can be of any type. However, as of v6.1, most (if not all) APIs using Record<P> are not generic: RecordReader, RecordWriter, etc:

public interface RecordReader {
   Record readRecord() throws Exception; // using raw type
   // open/close methods omitted
}

Using raw types in those key APIs has a few drawbacks:

• It is not type safe
• It requires additional casts in custom implementations of processors, mappers, writers, etc.
• It generates "raw types usage" warnings

The following snippet is a valid job definition in v6 that fails with a ClassCastException at runtime:

Job job = new JobBuilder()
        .reader(new IterableRecordReader(Arrays.asList("foo", "bar")))
        .processor(new RecordProcessor<Record<Integer>, Record<Integer>>() {
            @Override
            public Record<Integer> processRecord(Record<Integer> record) throws Exception {
                return new GenericRecord<>(record.getHeader(), record.getPayload() + 1); // ClassCastException
            }
        })
        .writer(new StandardOutputRecordWriter())
        .build();

JobExecutor jobExecutor = new JobExecutor();
JobReport report = jobExecutor.execute(job);
jobExecutor.shutdown();

All APIs should be updated to use generics for more type safety, and especially JobBuilder which should enforce that record types are as expected and coherent between the reader, processor and writer. With that, the previous job definition should not compile until we explicitly specify the input/output types. A type safe version of such a job definition would be something like:

Job job = new JobBuilder<Integer, Integer>() // Explicit specification of the input/output types
        .reader(new IterableRecordReader<>(Arrays.asList(1 , 2)))
        .processor(new RecordProcessor<Integer, Integer>() {
            @Override
            public Record<Integer> processRecord(Record<Integer> record) {
                return new GenericRecord<>(record.getHeader(), record.getPayload() + 1);
            }
        })
        .writer(new StandardOutputRecordWriter<>())
        .build();

[fmbenhassine]

The following APIs have been updated to implement this enhancement:

• RecordReader (and its listener): 7ed62a5
• Batch (and its listener): ba60977
• RecordWriter (and its listener): d437aa5
• RecordProcessor (and its listener / sub interfaces): 279ed29
• Predicate: 063fcc5
• JobBuilder: 6929936