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 testResetMissingPartitions() throws InvalidProtocolBufferException {
    HashMap<ByteStringRange, Instant> missingPartitionsDuration = new HashMap<>();
    missingPartitionsDuration.put(ByteStringRange.create("A", "B"), Instant.now());
    metadataTableDao.writeDetectNewPartitionMissingPartitions(missingPartitionsDuration);
    assertFalse(metadataTableDao.readDetectNewPartitionMissingPartitions().isEmpty());
    action.run(receiver);
    assertTrue(metadataTableDao.readDetectNewPartitionMissingPartitions().isEmpty());
  }