@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 testRestrictionClaimedAndIsSplitCase() {
    final String partitionToken = "partitionToken";
    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",
            Arrays.asList(
                new ChildPartition("childPartition1", partitionToken),
                new ChildPartition("childPartition2", partitionToken)),
            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("childPartition1")).thenReturn(null);
    when(transaction.getPartition("childPartition2")).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("childPartition1")
                .setParentTokens(Sets.newHashSet(partitionToken))
                .setStartTimestamp(startTimestamp)
                .setEndTimestamp(endTimestamp)
                .setHeartbeatMillis(heartbeat)
                .setState(CREATED)
                .setWatermark(startTimestamp)
                .build());
    verify(transaction)
        .insert(
            PartitionMetadata.newBuilder()
                .setPartitionToken("childPartition2")
                .setParentTokens(Sets.newHashSet(partitionToken))
                .setStartTimestamp(startTimestamp)
                .setEndTimestamp(endTimestamp)
                .setHeartbeatMillis(heartbeat)
                .setState(CREATED)
                .setWatermark(startTimestamp)
                .build());
  }