public EpochEntry findEpochEntryByOffset(final long offset) {
        this.readLock.lock();
        try {
            if (!this.epochMap.isEmpty()) {
                for (Map.Entry<Integer, EpochEntry> entry : this.epochMap.entrySet()) {
                    if (entry.getValue().getStartOffset() <= offset && entry.getValue().getEndOffset() > offset) {
                        return new EpochEntry(entry.getValue());
                    }
                }
            }
            return null;
        } finally {
            this.readLock.unlock();
        }
    }
@Test
    public void testFindEpochEntryByOffset() {
        final EpochEntry entry = this.epochCache.findEpochEntryByOffset(350);
        assertEquals(entry.getEpoch(), 2);
        assertEquals(entry.getStartOffset(), 300);
        assertEquals(entry.getEndOffset(), 500);
    }