code: boolean handleEndOfSegment(Segment segmentCompleted) throws ReaderNotInReaderGroupException {
          final Map<Segment, List<Long>> segmentToPredecessor;
          if (sync.getState().getEndSegments().containsKey(segmentCompleted)) {
              segmentToPredecessor = Collections.emptyMap();
          } else {
              val successors = getAndHandleExceptions(controller.getSuccessors(segmentCompleted), RuntimeException::new);
              segmentToPredecessor = successors.getSegmentToPredecessor();
          }
  
          AtomicBoolean reinitRequired = new AtomicBoolean(false);
          boolean result = sync.updateState((state, updates) -> {
              if (!state.isReaderOnline(readerId)) {
                  reinitRequired.set(true);
              } else {
                  log.debug("Marking segment {} as completed in reader group. CurrentState is: {}", segmentCompleted, state);
                  reinitRequired.set(false);
                  //This check guards against another checkpoint having started.
                  if (state.getCheckpointForReader(readerId) == null) {
                      updates.add(new SegmentCompleted(readerId, segmentCompleted, segmentToPredecessor));
                      return true;
                  }
              }
              return false;
          });
          if (reinitRequired.get()) {
              throw new ReaderNotInReaderGroupException(readerId);
          }
          acquireTimer.zero();
          return result;
      }
docs: Handles a segment being completed by calling the controller to gather all successors to the
  completed segment. To ensure consistent checkpoints, a segment cannot be released while a
  checkpoint for the reader is pending, so it may or may not succeed.
  
  @return true if the completed segment was released successfully.