@Override
    public int readFile(final ByteBufferCollector metaBufferCollector, final String fileName, final long offset,
                        final long maxCount) throws IOException, RetryAgainException {
        // read the whole meta file.
        if (fileName.equals(Snapshot.JRAFT_SNAPSHOT_META_FILE)) {
            final ByteBuffer metaBuf = this.metaTable.saveToByteBufferAsRemote();
            // because bufRef will flip the buffer before using, so we must set the meta buffer position to it's limit.
            BufferUtils.position(metaBuf, metaBuf.limit());
            metaBufferCollector.setBuffer(metaBuf);
            return EOF;
        }
        final LocalFileMeta fileMeta = this.metaTable.getFileMeta(fileName);
        if (fileMeta == null) {
            throw new FileNotFoundException("LocalFileMeta not found for " + fileName);
        }

        // go through throttle
        long newMaxCount = maxCount;
        if (this.snapshotThrottle != null) {
            newMaxCount = this.snapshotThrottle.throttledByThroughput(maxCount);
            if (newMaxCount < maxCount) {
                // if it's not allowed to read partly or it's allowed but
                // throughput is throttled to 0, try again.
                if (newMaxCount == 0) {
                    throw new RetryAgainException("readFile throttled by throughput");
                }
            }
        }

        return readFileWithMeta(metaBufferCollector, fileName, fileMeta, offset, newMaxCount);
    }
@Test
    public void testReadFile() throws Exception {
        final ByteBufferCollector bufRef = ByteBufferCollector.allocate();
        try {
            this.reader.readFile(bufRef, "unfound", 0, 1024);
            fail();
        } catch (final FileNotFoundException e) {

        }

        final String data = writeData();
        addDataMeta();

        final int read = this.reader.readFile(bufRef, "data", 0, 1024);
        assertEquals(-1, read);
        final ByteBuffer buf = bufRef.getBuffer();
        BufferUtils.flip(buf);
        assertEquals(data.length(), buf.remaining());
        final byte[] bs = new byte[data.length()];
        buf.get(bs);
        assertEquals(data, new String(bs));

    }