@Override
  public RecordReader<ImmutableBytesWritable, Result> createRecordReader(InputSplit split,
    TaskAttemptContext context) throws IOException, InterruptedException {
    TableSplit tSplit = (TableSplit) split;
    LOG.info(MessageFormat.format("Input split length: {0} bytes.", tSplit.getLength()));

    if (tSplit.getTable() == null) {
      throw new IOException("Cannot create a record reader because of a"
        + " previous error. Please look at the previous logs lines from"
        + " the task's full log for more details.");
    }
    final Connection connection = ConnectionFactory.createConnection(context.getConfiguration());
    Table table = connection.getTable(tSplit.getTable());

    if (this.tableRecordReader == null) {
      this.tableRecordReader = new TableRecordReader();
    }
    final TableRecordReader trr = this.tableRecordReader;

    try {
      Scan sc = tSplit.getScan();
      sc.withStartRow(tSplit.getStartRow());
      sc.withStopRow(tSplit.getEndRow());
      trr.setScan(sc);
      trr.setTable(table);
      return new RecordReader<ImmutableBytesWritable, Result>() {

        @Override
        public void close() throws IOException {
          trr.close();
          connection.close();
        }

        @Override
        public ImmutableBytesWritable getCurrentKey() throws IOException, InterruptedException {
          return trr.getCurrentKey();
        }

        @Override
        public Result getCurrentValue() throws IOException, InterruptedException {
          return trr.getCurrentValue();
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
          return trr.getProgress();
        }

        @Override
        public void initialize(InputSplit inputsplit, TaskAttemptContext context)
          throws IOException, InterruptedException {
          trr.initialize(inputsplit, context);
        }

        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
          return trr.nextKeyValue();
        }
      };
    } catch (IOException ioe) {
      // If there is an exception make sure that all
      // resources are closed and released.
      trr.close();
      connection.close();
      throw ioe;
    }
  }
Builds a TableRecordReader. If no TableRecordReader was provided, uses the default.
@param split   The split to work with.
@param context The current context.
@return The newly created record reader.
@throws IOException          When creating the reader fails.
@throws InterruptedException when record reader initialization fails
@see InputFormat#createRecordReader(InputSplit, TaskAttemptContext)
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/MultiTableInputFormatBase.java