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

    final String token = partition.getPartitionToken();

    LOG.debug("[{}] Processing child partition record {}", token, record);

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

    for (ChildPartition childPartition : record.getChildPartitions()) {
      processChildPartition(partition, record, childPartition);
    }

    LOG.debug("[{}] Child partitions action completed successfully", token);
    return Optional.empty();
  }
@Test
  public void testRestrictionClaimedAndIsMergeCaseAndChildNotExists() {
    final String partitionToken = "partitionToken";
    final String anotherPartitionToken = "anotherPartitionToken";
    final String childPartitionToken = "childPartition1";
    final HashSet<String> parentTokens = Sets.newHashSet(partitionToken, anotherPartitionToken);
    final long heartbeat = 30L;
    final Timestamp startTimestamp = Timestamp.ofTimeMicroseconds(10L);
    final Timestamp endTimestamp = Timestamp.ofTimeMicroseconds(20L);
    final PartitionMetadata partition = mock(PartitionMetadata.class);
    final ChildPartitionsRecord record =
        new ChildPartitionsRecord(
            startTimestamp,
            "recordSequence",
            Collections.singletonList(new ChildPartition(childPartitionToken, parentTokens)),
            null);
    when(partition.getEndTimestamp()).thenReturn(endTimestamp);
    when(partition.getHeartbeatMillis()).thenReturn(heartbeat);
    when(partition.getPartitionToken()).thenReturn(partitionToken);
    when(tracker.tryClaim(startTimestamp)).thenReturn(true);
    when(transaction.getPartition(childPartitionToken)).thenReturn(null);

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

    assertEquals(Optional.empty(), maybeContinuation);
    verify(watermarkEstimator).setWatermark(new Instant(startTimestamp.toSqlTimestamp().getTime()));
    verify(transaction)
        .insert(
            PartitionMetadata.newBuilder()
                .setPartitionToken(childPartitionToken)
                .setParentTokens(parentTokens)
                .setStartTimestamp(startTimestamp)
                .setEndTimestamp(endTimestamp)
                .setHeartbeatMillis(heartbeat)
                .setState(CREATED)
                .setWatermark(startTimestamp)
                .build());
  }