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

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

    final Timestamp timestamp = record.getTimestamp();
    final Instant timestampInstant = new Instant(timestamp.toSqlTimestamp().getTime());
    if (!tracker.tryClaim(timestamp)) {
      LOG.debug("[{}] Could not claim queryChangeStream({}), stopping", token, timestamp);
      return Optional.of(ProcessContinuation.stop());
    }
    metrics.incHeartbeatRecordCount();
    watermarkEstimator.setWatermark(timestampInstant);

    LOG.debug("[{}] Heartbeat record action completed successfully", token);
    return Optional.empty();
  }
@Test
  public void testRestrictionNotClaimed() {
    final String partitionToken = "partitionToken";
    final Timestamp timestamp = Timestamp.ofTimeMicroseconds(10L);

    when(tracker.tryClaim(timestamp)).thenReturn(false);
    when(partition.getPartitionToken()).thenReturn(partitionToken);

    final Optional<ProcessContinuation> maybeContinuation =
        action.run(partition, new HeartbeatRecord(timestamp, null), tracker, watermarkEstimator);

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