@Override
  public RecordReader<K, ProtoMessageWritable<V>> getRecordReader(InputSplit split,
      JobConf job, Reporter reporter) throws IOException {

    final Parser<V> parser = getParser(job.get(PROTO_CLASS));
    final RecordReader<K, ProtoMessageWritable<V>> reader = getSafeRecordReader(
        split, job, reporter);
    return new RecordReader<K, ProtoMessageWritable<V>>() {
      // Overload create value, since there is no default constructor for ProtoMessageWritable.
      @SuppressWarnings("unchecked")
      @Override
      public ProtoMessageWritable<V> createValue() {
        try {
          @SuppressWarnings("rawtypes")
          Constructor<ProtoMessageWritable> cons = ProtoMessageWritable.class
              .getDeclaredConstructor(Parser.class);
          cons.setAccessible(true);
          return cons.newInstance(parser);
        } catch (Exception e) {
          throw new RuntimeException("Unexpected error: ", e);
        }
      }

      @Override
      public K createKey() {
        return reader != null ? reader.createKey() : null;
      }

      @Override
      public void close() throws IOException {
        if (reader != null) {
          reader.close();
        }
      }

      @Override
      public long getPos() throws IOException {
        return reader != null ? reader.getPos() : 0;
      }

      @Override
      public float getProgress() throws IOException {
        return reader != null ? reader.getProgress() : 1.0f;
      }

      @Override
      public boolean next(K arg0, ProtoMessageWritable<V> arg1) throws IOException {
        return reader != null ? reader.next(arg0, arg1) : false;
      }
    };
  }
InputFormat to support reading ProtoWritable stored in a sequence file. You cannot use the
sequence file directly since the createValue method uses default constructor. But ProtoWritable
has a package protected constructor which takes a parser.
By reading the proto class name from job conf which is copied from table properties by Hive this
class manages to give a generic implementation where only can set the proto.class in table
properties and load the file.

It is also enhanced to ignore EOF exception while opening a file, so as to ignore 0 bytes files
in the table. Maybe we should allow this to be configured.

@param <K> K for the sequence file.
@param <V> The proto message type stored in the sequence file. Just to keep java compiler happy.
ql/src/java/org/apache/hadoop/hive/ql/io/protobuf/ProtobufMessageInputFormat.java