code: @Override
      public void markUnread(String channel, Collection<EventId> events) {
          // For each slab keep track of the earliest index for each unread event.
          ConcurrentMap<ChannelSlab, Integer> channelSlabs = Maps.newConcurrentMap();
          for (EventId event : events) {
              AstyanaxEventId astyanaxEvent = (AstyanaxEventId) event;
              checkArgument(channel.equals(astyanaxEvent.getChannel()));
              channelSlabs.merge(new ChannelSlab(channel, astyanaxEvent.getSlabId()), astyanaxEvent.getEventIdx(), Ints::min);
          }
  
          for (Map.Entry<ChannelSlab, Integer> entry : channelSlabs.entrySet()) {
              ChannelSlab channelSlab = entry.getKey();
              int eventIdx = entry.getValue();
              // Get the closed slab cursor, if any
              SlabCursor cursor = _closedSlabCursors.getIfPresent(channelSlab);
              // If the cursor exists and is beyond the lowest unread index, rewind it
              if (cursor != null && (cursor.get() == SlabCursor.END || cursor.get() > eventIdx)) {
                  // Synchronize on the cursor before updating it to avoid concurrent updates with a read
                  //noinspection SynchronizationOnLocalVariableOrMethodParameter
                  synchronized (cursor) {
                      if (cursor.get() == SlabCursor.END || cursor.get() > eventIdx) {
                          cursor.set(eventIdx);
                      }
                  }
              }
          }
      }
docs: When all events from a slab have been read via {@link #readNewer(String, EventSink)} _closedSlabCursors
  caches that the slab is empty for 10 seconds.  When a slab is marked as unread update the cursor, if present,
  such that it is rewound far enough to read the unread event again.