@ProcessElement
  public ProcessContinuation processElement(
      RestrictionTracker<OffsetByteRange, OffsetByteProgress> tracker,
      @Element SubscriptionPartition subscriptionPartition,
      OutputReceiver<SequencedMessage> receiver)
      throws Exception {
    LOG.debug("Starting process for {} at {}", subscriptionPartition, Instant.now());
    SubscriptionPartitionProcessor processor =
        processorFactory.newProcessor(subscriptionPartition, tracker, receiver);
    ProcessContinuation result = processor.run();
    LOG.debug("Starting commit for {} at {}", subscriptionPartition, Instant.now());
    // TODO(dpcollins-google): Move commits to a bundle finalizer for drain correctness
    processor
        .lastClaimed()
        .ifPresent(
            lastClaimed -> {
              try {
                committerFactory
                    .create(subscriptionPartition)
                    .commitOffset(Offset.of(lastClaimed.value() + 1));
              } catch (Exception e) {
                throw ExtractStatus.toCanonical(e).underlying;
              }
            });
    LOG.debug("Finishing process for {} at {}", subscriptionPartition, Instant.now());
    return result;
  }
@Test
  @SuppressWarnings("argument")
  public void process() throws Exception {
    when(processor.run()).thenReturn(ProcessContinuation.resume());
    when(processorFactory.newProcessor(any(), any(), any()))
        .thenAnswer(
            args -> {
              @Nonnull
              RestrictionTracker<OffsetRange, OffsetByteProgress> wrapped = args.getArgument(1);
              when(tracker.tryClaim(any())).thenReturn(true).thenReturn(false);
              assertTrue(wrapped.tryClaim(OffsetByteProgress.of(example(Offset.class), 123)));
              assertFalse(wrapped.tryClaim(OffsetByteProgress.of(Offset.of(333333), 123)));
              return processor;
            });
    doReturn(Optional.of(example(Offset.class))).when(processor).lastClaimed();
    assertEquals(ProcessContinuation.resume(), sdf.processElement(tracker, PARTITION, output));
    verify(processorFactory).newProcessor(eq(PARTITION), any(), eq(output));
    InOrder order = inOrder(processor);
    order.verify(processor).run();
    order.verify(processor).lastClaimed();
    InOrder order2 = inOrder(committerFactory, committer);
    order2.verify(committerFactory).create(PARTITION);
    order2.verify(committer).commitOffset(Offset.of(example(Offset.class).value() + 1));
  }