public ProcessContinuation run(
      RestrictionTracker<TimestampRange, Timestamp> tracker,
      OutputReceiver<PartitionMetadata> receiver,
      ManualWatermarkEstimator<Instant> watermarkEstimator) {

    final Timestamp readTimestamp = tracker.currentRestriction().getFrom();
    // Updates the current watermark as the min of the watermarks from all existing partitions
    final Timestamp minWatermark = dao.getUnfinishedMinWatermark();

    if (minWatermark != null) {
      return processPartitions(tracker, receiver, watermarkEstimator, minWatermark, readTimestamp);
    } else {
      return terminate(tracker);
    }
  }
@Test
  public void testTerminatesWhenAllPartitionsAreFinished() {
    final Timestamp from = Timestamp.ofTimeMicroseconds(10L);
    when(restriction.getFrom()).thenReturn(from);
    when(dao.getUnfinishedMinWatermark()).thenReturn(null);

    final ProcessContinuation continuation = action.run(tracker, receiver, watermarkEstimator);

    assertEquals(ProcessContinuation.stop(), continuation);
    verify(watermarkEstimator, never()).setWatermark(any());
    verify(receiver, never()).outputWithTimestamp(any(), any());
  }