@Override
    public int getTailOffset() {
      if (!getOffsetMap().containsOffset(TAIL_OFFSET) && invalidateTrace != null) {
        throw new RuntimeExceptionWithAttachments("Tail offset invalid", new Attachment("invalidated", invalidateTrace));
      }

      int offset = super.getTailOffset();
      if (tailWatcher.getStartOffset() != tailWatcher.getEndOffset() && offset > 0) {
        watchTail(offset);
      }

      return offset;
    }
@Override
    public int getTailOffset()