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 testOutputNewPartitions() throws InvalidProtocolBufferException {
    ByteStringRange partition1 = ByteStringRange.create("A", "B");
    ChangeStreamContinuationToken token1 =
        ChangeStreamContinuationToken.create(partition1, "tokenAB");
    Instant watermark1 = Instant.now().minus(Duration.standardSeconds(10));
    NewPartition newSplitPartition =
        new NewPartition(partition1, Collections.singletonList(token1), watermark1);
    metadataTableDao.writeNewPartition(newSplitPartition);

    ByteStringRange partition2 = ByteStringRange.create("B", "D");
    ByteStringRange partition2parent1 = ByteStringRange.create("C", "D");
    ChangeStreamContinuationToken token2 =
        ChangeStreamContinuationToken.create(partition2parent1, "tokenCD");
    Instant watermark2 = Instant.now().plus(Duration.standardSeconds(10));
    NewPartition newPartitionMissingParent =
        new NewPartition(partition2, Collections.singletonList(token2), watermark2);
    metadataTableDao.writeNewPartition(newPartitionMissingParent);

    ByteStringRange partition3 = ByteStringRange.create("D", "E");
    ChangeStreamContinuationToken token3 =
        ChangeStreamContinuationToken.create(partition3, "tokenDE");
    Instant watermark3 = Instant.now().plus(Duration.standardSeconds(5));
    NewPartition deletedNewPartition =
        new NewPartition(partition3, Collections.singletonList(token3), watermark3);
    metadataTableDao.writeNewPartition(deletedNewPartition);
    metadataTableDao.markNewPartitionForDeletion(deletedNewPartition);

    // There are only 2 NewPartition rows because partition3 is deleted.
    assertEquals(2, metadataTableDao.readNewPartitions().size());

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

    PartitionRecord actualPartition1 = actualPartitions.get(0);
    assertEquals(partition1, actualPartition1.getPartition());
    assertEquals(1, actualPartition1.getChangeStreamContinuationTokens().size());
    assertEquals(token1, actualPartition1.getChangeStreamContinuationTokens().get(0));
    assertNull(actualPartition1.getStartTime());
    assertEquals(watermark1, actualPartition1.getParentLowWatermark());
    assertEquals(endTime, actualPartition1.getEndTime());
    assertEquals(
        Collections.singletonList(newSplitPartition), actualPartition1.getParentPartitions());
    // Uuid is filled.
    assertFalse(actualPartition1.getUuid().isEmpty());

    // The 2nd partition is not partition2 but partition3 because partition2 is missing a parent.
    // Even though partition3 is marked for deletion, we still process it.
    PartitionRecord actualPartition2 = actualPartitions.get(1);
    assertEquals(partition3, actualPartition2.getPartition());
    assertEquals(1, actualPartition2.getChangeStreamContinuationTokens().size());
    assertEquals(token3, actualPartition2.getChangeStreamContinuationTokens().get(0));
    assertNull(actualPartition2.getStartTime());
    assertEquals(watermark3, actualPartition2.getParentLowWatermark());
    assertEquals(endTime, actualPartition2.getEndTime());
    assertEquals(
        Collections.singletonList(deletedNewPartition), actualPartition2.getParentPartitions());

    // There is only 1 NewPartition row now because we processed them except for the NewPartition
    // with missing parent.
    assertEquals(1, metadataTableDao.readNewPartitions().size());
  }