@ProcessElement
  public void processElement(OutputReceiver<InitialPipelineState> receiver) throws IOException {
    LOG.info(daoFactory.getStreamTableDebugString());
    LOG.info(daoFactory.getMetadataTableDebugString());
    LOG.info("ChangeStreamName: " + daoFactory.getChangeStreamName());

    boolean resume = false;
    DetectNewPartitionsState detectNewPartitionsState =
        daoFactory.getMetadataTableDao().readDetectNewPartitionsState();

    switch (existingPipelineOptions) {
      case RESUME_OR_NEW:
        // perform resumption.
        if (detectNewPartitionsState != null) {
          resume = true;
          startTime = detectNewPartitionsState.getWatermark();
          LOG.info("Resuming from previous pipeline with low watermark of {}", startTime);
        } else {
          LOG.info(
              "Attempted to resume, but previous watermark does not exist, starting at {}",
              startTime);
        }
        break;
      case RESUME_OR_FAIL:
        // perform resumption.
        if (detectNewPartitionsState != null) {
          resume = true;
          startTime = detectNewPartitionsState.getWatermark();
          LOG.info("Resuming from previous pipeline with low watermark of {}", startTime);
        } else {
          LOG.error("Previous pipeline with the same change stream name doesn't exist, stopping");
          return;
        }
        break;
      case FAIL_IF_EXISTS:
        if (detectNewPartitionsState != null) {
          LOG.error(
              "A previous pipeline exists with the same change stream name and existingPipelineOption is set to FAIL_IF_EXISTS.");
          return;
        }
        break;
      case SKIP_CLEANUP:
        if (detectNewPartitionsState != null) {
          LOG.error(
              "A previous pipeline exists with the same change stream name and existingPipelineOption is set to SKIP_CLEANUP. This option should only be used in tests.");
          return;
        }
        break;
      default:
        LOG.error("Unexpected existingPipelineOptions option.");
        // terminate pipeline
        return;
    }
    daoFactory.getMetadataTableDao().writeDetectNewPartitionVersion();
    receiver.output(new InitialPipelineState(startTime, resume));
  }
@Test
  public void testInitializeSkipCleanupWithoutDNP() throws IOException {
    ByteString metadataRowKey =
        metadataTableAdminDao
            .getChangeStreamNamePrefix()
            .concat(ByteString.copyFromUtf8("existing_row"));
    dataClient.mutateRow(
        RowMutation.create(tableId, metadataRowKey)
            .setCell(
                MetadataTableAdminDao.CF_WATERMARK, MetadataTableAdminDao.QUALIFIER_DEFAULT, 123));
    Instant startTime = Instant.now();
    InitializeDoFn initializeDoFn =
        new InitializeDoFn(daoFactory, startTime, ExistingPipelineOptions.SKIP_CLEANUP);
    initializeDoFn.processElement(outputReceiver);
    // Skip cleanup will always resume from startTime
    verify(outputReceiver, times(1)).output(new InitialPipelineState(startTime, false));
    // Existing metadata shouldn't be cleaned up
    assertNotNull(dataClient.readRow(tableId, metadataRowKey));
  }