public boolean readRecord(Record record) throws IOException, InterruptedException {
        boolean hasNext = false;
        hasNext = this.phoenixRecordReader.nextKeyValue();
        if (!hasNext)
            return hasNext;
        PhoenixRecordWritable phoenixRecordWritable =  (PhoenixRecordWritable) this.phoenixRecordReader.getCurrentValue();
        Map<String, Object> phoenixRecord = phoenixRecordWritable.getResultMap();
        this.constructRecordFromPhoenix(record, phoenixRecord);
        return hasNext;
    }
@Test
    public void testReadRecord() throws Exception {
        List<Configuration> splits = this.generateSplitConfig();

        int allRecordNum = 0;
        for (int i = 0; i < splits.size(); i++) {
            RecordSender recordSender = mock(RecordSender.class);
            when(recordSender.createRecord()).thenReturn(new DefaultRecord());
            Record record = recordSender.createRecord();

            HbaseSQLReaderTask hbase11SQLReaderTask = new HbaseSQLReaderTask(splits.get(i));
            hbase11SQLReaderTask.init();
            hbase11SQLReaderTask.prepare();

            int num = 0;
            while (true) {
                boolean hasLine = false;
                try {
                    hasLine = hbase11SQLReaderTask.readRecord(record);
                } catch (Exception e) {
                    e.printStackTrace();
                    throw e;
                }
                if (!hasLine)
                    break;
                num++;
                if (num % 100 == 0)
                    System.out.println("record num is :" + num + ",record is " + record.toString());
                when(recordSender.createRecord()).thenReturn(new DefaultRecord());
                String recordStr = "";
                for (int j = 0; j < record.getColumnNumber(); j++) {
                    recordStr += record.getColumn(j).asString() + ",";
                }
                recordSender.sendToWriter(record);
                record = recordSender.createRecord();
            }
            System.out.println("split id is " + i + ",record num = " + num);
            allRecordNum += num;
            recordSender.flush();
            hbase11SQLReaderTask.destroy();
        }
        System.out.println("all record num = " + allRecordNum);
        assertEquals(10000, allRecordNum);
    }