public CompositeDataBuffer getMultipleDataFileChannel(Channel channel, long length)
      throws IOException {
    if (mFileSize <= mPos) {
      // TODO(JiamingMai): consider throwing exception directly
      return new CompositeDataBuffer(Collections.emptyList());
    }
    List<DataBuffer> dataBufferList = new ArrayList<>();
    long bytesToTransfer = Math.min(length, mFileSize - mPos);
    long bytesToTransferLeft = bytesToTransfer;
    while (bytesToTransferLeft > 0) {
      long lengthPerOp = Math.min(bytesToTransferLeft, mPositionReader.getPageSize());
      DataBuffer dataBuffer;
      Optional<DataFileChannel> dataFileChannel =
          mPositionReader.getDataFileChannel(mPos, (int) lengthPerOp);
      if (!dataFileChannel.isPresent()) {
        dataBuffer = getDataBufferByCopying(channel, (int) lengthPerOp);
      } else {
        // update mPos
        // TODO(JiamingMai): need to lock page files since the openFile op is called in netty latter
        dataBuffer = dataFileChannel.get();
        if (dataBuffer.getLength() > 0) {
          mPos += dataBuffer.getLength();
        } else {
          dataBuffer = getDataBufferByCopying(channel, (int) lengthPerOp);
        }
      }
      // update bytesToTransferLeft
      bytesToTransferLeft -= dataBuffer.getLength();
      dataBufferList.add(dataBuffer);
    }
    CompositeDataBuffer compositeDataBuffer = new CompositeDataBuffer(dataBufferList);
    return compositeDataBuffer;
  }
@Test
  public void getMultipleDataFileChannel() throws IOException {
    CompositeDataBuffer compositeDataBuffer =
        mPagedFileReader.getMultipleDataFileChannel(mEmbeddedChannel, mFileLen);
    Assert.assertEquals(mFileLen, compositeDataBuffer.getLength());

    byte[] bArray = new byte[mFileLen];
    int readPosition = 0;
    List<DataBuffer> listDataBuffer = (List<DataBuffer>) compositeDataBuffer.getNettyOutput();
    for (DataBuffer dataBuffer : listDataBuffer) {
      int byteToBeRead = dataBuffer.readableBytes();
      dataBuffer.readBytes(bArray, readPosition, byteToBeRead);
      readPosition += byteToBeRead;
    }
    Assert.assertArrayEquals(mTestData, bArray);
  }