@Override
  public boolean advance() throws IOException {
    if (!subscriber.state().equals(State.RUNNING)) {
      Throwable t = subscriber.failureCause();
      if ("DUPLICATE_SUBSCRIBER_CONNECTIONS"
          .equals(ExtractStatus.getErrorInfoReason(ExtractStatus.toCanonical(t)))) {
        throw new IOException(
            "Partition reassigned to a different worker- this is expected and can be ignored.", t);
      }
      throw new IOException("Subscriber failed when trying to advance.", t);
    }
    if (advanced) {
      subscriber.pop();
    }
    Optional<SequencedMessage> next = subscriber.peek();
    advanced = next.isPresent();
    if (!advanced) {
      return false;
    }
    Offset nextOffset = Offset.of(next.get().getCursor().getOffset() + 1);
    checkState(nextOffset.value() > fetchOffset.value());
    fetchOffset = nextOffset;
    lastMessageTimestamp = Optional.of(getTimestamp(next.get()));
    return true;
  }
@Test
  public void advanceNoPreviousValue() throws Exception {
    startSubscriber();
    SequencedMessage message = messageWith("abc", 2, Instant.now());
    doReturn(Optional.of(message)).when(subscriber).peek();
    assertTrue(reader.advance());
    verify(subscriber, times(0)).pop();
  }