public void deleteWorkflowData(String workflowId, long internalId, long timeoutInNanos) {
    final long startNanos = System.nanoTime();
    Stage stage = getWorkflowDeletionStage(workflowId, internalId);

    int totalDeleted = 0;
    int deleted = 0;
    long stageTime = startNanos;
    Timeline timeline = new Timeline(null);

    while (stage != Stage.DELETION_DONE) {
      int cnt = deleteDataForStage(workflowId, internalId, stage);
      TimeUtils.sleep(BATCH_DELAY_IN_MILLIS)
          .ifPresent(
              details ->
                  LOG.info(
                      "Thread is interrupted, ignore it and just continue without sleeping: {}",
                      details));
      totalDeleted += cnt;
      deleted += cnt;
      long curTime = System.nanoTime();
      if (cnt < Constants.BATCH_DELETION_LIMIT || curTime - startNanos >= timeoutInNanos) {
        TimelineEvent event =
            TimelineLogEvent.info(
                "Deleted [%s] items in the stage of [%s], taking [%s] millis",
                deleted, stage.name(), TimeUnit.NANOSECONDS.toMillis(curTime - stageTime));
        LOG.info(event.getMessage());
        timeline.add(event);
        if (cnt < Constants.BATCH_DELETION_LIMIT) {
          deleted = 0;
          stage = DELETION_STAGES[stage.ordinal() + 1]; // move to the next stage
          stageTime = curTime;
        }
        if (curTime - startNanos >= timeoutInNanos) {
          break;
        }
      }
    }

    LOG.info(
        "Totally deleted {} rows while deleting workflow [{}][{}] and then "
            + "complete this round of deletion at stage [{}], taking [{}] millis.",
        totalDeleted,
        workflowId,
        internalId,
        stage.name(),
        TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos));

    int updated = updateWorkflowDeletionStage(workflowId, internalId, stage, timeline);
    if (updated != SUCCESS_WRITE_SIZE) {
      throw new MaestroRetryableError(
          "Failed to update workflow deletion data for workflow [%s][%s], retry it later",
          workflowId, internalId);
    }
    if (stage != Stage.DELETION_DONE) {
      LOG.info(
          "Workflow [{}][{}] deletion is timed out and will retry it later",
          workflowId,
          internalId);
      throw new MaestroTimeoutException(
          "Workflow [%s][%s] deletion is timed out", workflowId, internalId);
    }
  }
@Test
  public void testDeleteWorkflowData() throws Exception {
    WorkflowDefinition wfd = loadWorkflow(TEST_WORKFLOW_ID1);
    workflowDao.addWorkflowDefinition(wfd, wfd.getPropertiesSnapshot().extractProperties());
    reset(publisher);
    ArgumentCaptor<DeleteWorkflowJobEvent> argumentCaptor =
        ArgumentCaptor.forClass(DeleteWorkflowJobEvent.class);
    workflowDao.deleteWorkflow(TEST_WORKFLOW_ID1, User.create("tester"));
    Mockito.verify(publisher, times(1)).publishOrThrow(argumentCaptor.capture(), any());

    DeleteWorkflowJobEvent deleteWorkflowJobEvent = argumentCaptor.getValue();
    assertEquals(TEST_WORKFLOW_ID1, deleteWorkflowJobEvent.getWorkflowId());
    assertEquals("tester", deleteWorkflowJobEvent.getAuthor().getName());
    assertTrue(
        deletionDao.isDeletionInitialized(
            TEST_WORKFLOW_ID1, deleteWorkflowJobEvent.getInternalId()));

    deletionDao.deleteWorkflowData(
        TEST_WORKFLOW_ID1, deleteWorkflowJobEvent.getInternalId(), TimeUnit.MINUTES.toNanos(1));
    assertFalse(
        deletionDao.isDeletionInitialized(
            TEST_WORKFLOW_ID1, deleteWorkflowJobEvent.getInternalId()));
    assertFalse(deletionDao.isDeletionInProgress(TEST_WORKFLOW_ID1));
  }