@Override
  public ProcessContinuation run() {
    // Read any available data.
    for (Optional<SequencedMessage> next = subscriber.peek();
        next.isPresent();
        next = subscriber.peek()) {
      SequencedMessage message = next.get();
      Offset messageOffset = Offset.of(message.getCursor().getOffset());
      if (tracker.tryClaim(OffsetByteProgress.of(messageOffset, message.getSizeBytes()))) {
        subscriber.pop();
        lastClaimedOffset = Optional.of(messageOffset);
        receiver.outputWithTimestamp(
            message, new Instant(Timestamps.toMillis(message.getPublishTime())));
      } else {
        // Our claim failed, return stop()
        return ProcessContinuation.stop();
      }
    }
    // There is no more data available, yield to the runtime.
    return ProcessContinuation.resume();
  }
@Test
  public void create() {
    SubscriptionPartitionProcessor processor = newProcessor();
    assertEquals(ProcessContinuation.resume(), processor.run());
    InOrder order = inOrder(subscriber);
    order.verify(subscriber).fetchOffset();
    order.verify(subscriber).rebuffer();
  }