@VisibleForTesting
  public Optional<ProcessContinuation> run(
      PartitionMetadata partition,
      DataChangeRecord record,
      RestrictionTracker<TimestampRange, Timestamp> tracker,
      OutputReceiver<DataChangeRecord> outputReceiver,
      ManualWatermarkEstimator<Instant> watermarkEstimator) {

    final String token = partition.getPartitionToken();
    LOG.debug("[{}] Processing data record {}", token, record.getCommitTimestamp());

    final Timestamp commitTimestamp = record.getCommitTimestamp();
    final Instant commitInstant = new Instant(commitTimestamp.toSqlTimestamp().getTime());
    if (!tracker.tryClaim(commitTimestamp)) {
      LOG.debug("[{}] Could not claim queryChangeStream({}), stopping", token, commitTimestamp);
      return Optional.of(ProcessContinuation.stop());
    }
    outputReceiver.outputWithTimestamp(record, commitInstant);
    watermarkEstimator.setWatermark(commitInstant);

    throughputEstimator.update(Timestamp.now(), record);

    LOG.debug("[{}] Data record action completed successfully", token);
    return Optional.empty();
  }
@Test
  public void testRestrictionNotClaimed() {
    final String partitionToken = "partitionToken";
    final Timestamp timestamp = Timestamp.ofTimeMicroseconds(10L);
    final DataChangeRecord record = mock(DataChangeRecord.class);
    when(record.getCommitTimestamp()).thenReturn(timestamp);
    when(tracker.tryClaim(timestamp)).thenReturn(false);
    when(partition.getPartitionToken()).thenReturn(partitionToken);

    final Optional<ProcessContinuation> maybeContinuation =
        action.run(partition, record, tracker, outputReceiver, watermarkEstimator);

    assertEquals(Optional.of(ProcessContinuation.stop()), maybeContinuation);
    verify(outputReceiver, never()).outputWithTimestamp(any(), any());
    verify(watermarkEstimator, never()).setWatermark(any());
    verify(throughputEstimator, never()).update(any(), any());
  }