@Override
  public <T> T read(long recordId,
                    @NotNull ByteBufferReader<T> reader) throws IOException {
    long recordOffsetInFile = recordIdToOffset(recordId);

    //RC: Records between 'committed' and 'allocated' cursors are not all fully written -> we permit
    //    reading the records that are still not fully commited. This is good from API-consistency
    //    PoV: if .append() returns an id -- that id must be valid for .read(id), even though some
    //    records _before_ that id are not yet finished.

    long recordsAllocatedUpTo = firstUnAllocatedOffset();
    if (recordOffsetInFile >= recordsAllocatedUpTo) {
      throw new IllegalArgumentException(
        "Can't read recordId(=" + recordId + ", offset: " + recordOffsetInFile + "]: " +
        "outside of allocated region [<" + recordsAllocatedUpTo + "] " +
        moreDiagnosticInfo(recordOffsetInFile));
    }


    MMappedFileStorage.Page page = storage.pageByOffset(recordOffsetInFile);
    int recordOffsetInPage = storage.toOffsetInPage(recordOffsetInFile);
    ByteBuffer pageBuffer = page.rawPageBuffer();

    int recordHeader = RecordLayout.readHeader(pageBuffer, recordOffsetInPage);
    if (!RecordLayout.isDataHeader(recordHeader)) {
      throw new IOException("record[" + recordId + "][@" + recordOffsetInFile + "] is " +
                            "PaddingRecord(header=" + Integer.toHexString(recordHeader) + ") -- i.e. has no data. " +
                            moreDiagnosticInfo(recordOffsetInFile)
      );
    }
    if (!RecordLayout.isRecordCommitted(recordHeader)) {
      throw new IOException("record[" + recordId + "][@" + recordOffsetInFile + "] is not commited: " +
                            "(header=" + Integer.toHexString(recordHeader) + ") either not yet written or corrupted. " +
                            moreDiagnosticInfo(recordOffsetInFile) +
                            (APPEND_LOG_DUMP_ON_ERROR
                             ? "\n" +
                               dumpContentAroundId(recordId, DEBUG_DUMP_REGION_WIDTH, MAX_RECORD_SIZE_TO_DUMP)
                             : "")
      );
    }
    int payloadLength = RecordLayout.extractPayloadLength(recordHeader);
    if (!RecordLayout.isFitIntoPage(pageBuffer, recordOffsetInPage, payloadLength)) {
      throw new IOException("record[" + recordId + "][@" + recordOffsetInFile + "].payloadLength(=" + payloadLength + ") " +
                            "is incorrect: page[0.." + pageBuffer.limit() + "], " +
                            "committedUpTo: " + firstUnCommittedOffset() + ", allocatedUpTo: " + firstUnAllocatedOffset() + ". " +
                            moreDiagnosticInfo(recordOffsetInFile) +
                            (APPEND_LOG_DUMP_ON_ERROR
                             ? "\n" +
                               dumpContentAroundId(recordId, DEBUG_DUMP_REGION_WIDTH, MAX_RECORD_SIZE_TO_DUMP)
                             : "")
      );
    }
    ByteBuffer recordDataSlice = pageBuffer.slice(recordOffsetInPage + RecordLayout.DATA_OFFSET, payloadLength)
      //.asReadOnlyBuffer()
      .order(pageBuffer.order());
    return reader.read(recordDataSlice);
  }
@Override
  public <T> T read(long recordId,
                    @NotNull ByteBufferReader<T> reader)