@ProcessElement
  public ProcessContinuation processElement(
      @Element PartitionMetadata partition,
      RestrictionTracker<TimestampRange, com.google.cloud.Timestamp> tracker,
      OutputReceiver<DataChangeRecord> receiver,
      ManualWatermarkEstimator<Instant> watermarkEstimator,
      BundleFinalizer bundleFinalizer) {

    final String token = partition.getPartitionToken();

    LOG.debug("[{}] Processing element with restriction {}", token, tracker.currentRestriction());

    return queryChangeStreamAction.run(
        partition, tracker, receiver, watermarkEstimator, bundleFinalizer);
  }
@Test
  public void testQueryChangeStreamMode() {
    when(queryChangeStreamAction.run(any(), any(), any(), any(), any()))
        .thenReturn(ProcessContinuation.stop());

    final ProcessContinuation result =
        doFn.processElement(partition, tracker, receiver, watermarkEstimator, bundleFinalizer);

    assertEquals(ProcessContinuation.stop(), result);
    verify(queryChangeStreamAction)
        .run(partition, tracker, receiver, watermarkEstimator, bundleFinalizer);

    verify(dataChangeRecordAction, never()).run(any(), any(), any(), any(), any());
    verify(heartbeatRecordAction, never()).run(any(), any(), any(), any());
    verify(childPartitionsRecordAction, never()).run(any(), any(), any(), any());
    verify(tracker, never()).tryClaim(any());
  }