public void run(OutputReceiver<PartitionRecord> receiver, Instant startTime) {
    List<ByteStringRange> streamPartitions =
        changeStreamDao.generateInitialChangeStreamPartitions();
    for (ByteStringRange partition : streamPartitions) {
      metrics.incListPartitionsCount();
      String uid = UniqueIdGenerator.getNextId();
      PartitionRecord partitionRecord =
          new PartitionRecord(
              partition, startTime, uid, startTime, Collections.emptyList(), endTime);
      // We are outputting elements with timestamp of 0 to prevent reliance on event time. This
      // limits the ability to window on commit time of any data changes. It is still possible to
      // window on processing time.
      receiver.outputWithTimestamp(partitionRecord, Instant.EPOCH);
    }
  }
@Test
  public void testGenerateInitialPartitionsFromStartTime() {
    Range.ByteStringRange partition1 = Range.ByteStringRange.create("", "b");
    Range.ByteStringRange partition2 = Range.ByteStringRange.create("b", "");
    List<Range.ByteStringRange> partitionRecordList = Arrays.asList(partition1, partition2);
    when(changeStreamDao.generateInitialChangeStreamPartitions()).thenReturn(partitionRecordList);

    GenerateInitialPartitionsAction generateInitialPartitionsAction =
        new GenerateInitialPartitionsAction(metrics, changeStreamDao, endTime);
    generateInitialPartitionsAction.run(receiver, startTime);
    verify(receiver, times(2))
        .outputWithTimestamp(partitionRecordArgumentCaptor.capture(), eq(Instant.EPOCH));
    List<PartitionRecord> actualPartitions = partitionRecordArgumentCaptor.getAllValues();

    assertEquals(partition1, actualPartitions.get(0).getPartition());
    assertEquals(startTime, actualPartitions.get(0).getStartTime());
    assertEquals(partition2, actualPartitions.get(1).getPartition());
    assertEquals(startTime, actualPartitions.get(1).getStartTime());
  }