public void run(OutputReceiver<PartitionRecord> receiver) throws InvalidProtocolBufferException {
    // Erase any existing missing partitions.
    metadataTableDao.writeDetectNewPartitionMissingPartitions(new HashMap<>());

    List<PartitionRecord> partitions = metadataTableDao.readAllStreamPartitions();

    for (PartitionRecord partitionRecord : partitions) {
      if (partitionRecord.getUuid().isEmpty()) {
        partitionRecord.setUuid(UniqueIdGenerator.getNextId());
      }
      if (endTime != null) {
        partitionRecord.setEndTime(endTime);
      }
      LOG.info("DNP: Outputting existing partition: {}", partitionRecord);
      metrics.incListPartitionsCount();
      receiver.outputWithTimestamp(partitionRecord, Instant.EPOCH);
    }

    List<NewPartition> newPartitions = metadataTableDao.readNewPartitionsIncludingDeleted();

    for (NewPartition newPartition : newPartitions) {
      processNewPartitionsAction.processNewPartition(newPartition, receiver);
    }
  }
@Test
  public void testOutputExistingStreamPartitions() throws InvalidProtocolBufferException {
    ByteStringRange partition1 = ByteStringRange.create("A", "B");
    Instant watermark1 = Instant.now().minus(Duration.standardSeconds(10));
    PartitionRecord partitionWithStartTime =
        new PartitionRecord(partition1, watermark1, "1", watermark1, Collections.emptyList(), null);
    metadataTableDao.lockAndRecordPartition(partitionWithStartTime);

    ByteStringRange partition2 = ByteStringRange.create("B", "D");
    ChangeStreamContinuationToken partition2Token1 =
        ChangeStreamContinuationToken.create(ByteStringRange.create("B", "C"), "tokenBC");
    ChangeStreamContinuationToken partition2Token2 =
        ChangeStreamContinuationToken.create(ByteStringRange.create("C", "D"), "tokenCD");
    Instant watermark2 = Instant.now().plus(Duration.standardMinutes(1));
    PartitionRecord partitionWithInitialTokens =
        new PartitionRecord(
            partition2,
            Arrays.asList(partition2Token1, partition2Token2),
            "2",
            watermark2,
            Collections.emptyList(),
            null);
    metadataTableDao.lockAndRecordPartition(partitionWithInitialTokens);

    ByteStringRange partition3 = ByteStringRange.create("D", "H");
    Instant watermark3 = Instant.now();
    PartitionRecord partitionWithContinuationToken =
        new PartitionRecord(partition3, watermark3, "3", watermark3, Collections.emptyList(), null);
    metadataTableDao.lockAndRecordPartition(partitionWithContinuationToken);
    ChangeStreamContinuationToken partition3token =
        ChangeStreamContinuationToken.create(ByteStringRange.create("D", "H"), "tokenDH");
    metadataTableDao.updateWatermark(partition3, watermark3, partition3token);

    ByteStringRange partition4 = ByteStringRange.create("H", "I");
    ChangeStreamContinuationToken partition4token =
        ChangeStreamContinuationToken.create(ByteStringRange.create("H", "I"), "tokenHI");
    Instant watermark4 = Instant.now();
    PartitionRecord unlockedPartition =
        new PartitionRecord(
            partition4,
            Collections.singletonList(partition4token),
            "4",
            watermark4,
            Collections.emptyList(),
            null);
    metadataTableDao.lockAndRecordPartition(unlockedPartition);
    metadataTableDao.releaseStreamPartitionLockForDeletion(partition4, unlockedPartition.getUuid());

    action.run(receiver);
    verify(receiver, times(4))
        .outputWithTimestamp(partitionRecordArgumentCaptor.capture(), eq(Instant.EPOCH));
    List<PartitionRecord> actualPartitions = partitionRecordArgumentCaptor.getAllValues();

    PartitionRecord actualPartition1 = actualPartitions.get(0);
    assertEquals(partition1, actualPartition1.getPartition());
    assertEquals(partitionWithStartTime.getUuid(), actualPartition1.getUuid());
    assertNull(actualPartition1.getChangeStreamContinuationTokens());
    assertEquals(partitionWithStartTime.getStartTime(), actualPartition1.getStartTime());
    assertEquals(endTime, actualPartition1.getEndTime());
    assertEquals(
        partitionWithStartTime.getParentLowWatermark(), actualPartition1.getParentLowWatermark());

    PartitionRecord actualPartition2 = actualPartitions.get(1);
    assertEquals(partition2, actualPartition2.getPartition());
    assertEquals(partitionWithInitialTokens.getUuid(), actualPartition2.getUuid());
    assertNull(actualPartition2.getStartTime());
    assertEquals(
        partitionWithInitialTokens.getChangeStreamContinuationTokens(),
        actualPartition2.getChangeStreamContinuationTokens());
    assertEquals(endTime, actualPartition2.getEndTime());
    assertEquals(
        partitionWithInitialTokens.getParentLowWatermark(),
        actualPartition2.getParentLowWatermark());

    PartitionRecord actualPartition3 = actualPartitions.get(2);
    assertEquals(partition3, actualPartition3.getPartition());
    assertEquals(partitionWithContinuationToken.getUuid(), actualPartition3.getUuid());
    assertNull(actualPartition3.getStartTime());
    assertNotNull(actualPartition3.getChangeStreamContinuationTokens());
    assertEquals(1, actualPartition3.getChangeStreamContinuationTokens().size());
    assertEquals(partition3token, actualPartition3.getChangeStreamContinuationTokens().get(0));
    assertEquals(endTime, actualPartition3.getEndTime());
    assertEquals(
        partitionWithContinuationToken.getParentLowWatermark(),
        actualPartition3.getParentLowWatermark());

    PartitionRecord actualPartition4 = actualPartitions.get(3);
    assertEquals(partition4, actualPartition4.getPartition());
    // partition4 was unlocked so it gets a new uuid.
    assertNotEquals("4", actualPartition4.getUuid());
    assertNull(actualPartition4.getStartTime());
    assertNotNull(actualPartition4.getChangeStreamContinuationTokens());
    assertEquals(1, actualPartition4.getChangeStreamContinuationTokens().size());
    assertEquals(partition4token, actualPartition4.getChangeStreamContinuationTokens().get(0));
    assertEquals(endTime, actualPartition4.getEndTime());
    assertEquals(
        unlockedPartition.getParentLowWatermark(), actualPartition4.getParentLowWatermark());
  }