@Override
  public List<Task> createTasks(List<Task> tasks) {
    validateTasks(tasks);
    Collection<Task> pendingTasks =
        tasks.stream()
            .collect(
                Collectors.toMap(Task::getTaskId, Function.identity(), (u, v) -> v, TreeMap::new))
            .values();

    return withMetricLogError(
        () -> {
          int cnt =
              withRetryableStatement(
                  CREATE_TASK_STATEMENT,
                  statement -> {
                    int count = 0;
                    int inserted = 0;
                    for (Task task : pendingTasks) {
                      addTask(statement, task);
                      statement.addBatch();
                      count++;
                      // batch INSERT with a batch size
                      if (count % insertBatchSize == 0 || count == pendingTasks.size()) {
                        int[] res = statement.executeBatch();
                        inserted += res.length;
                      }
                    }
                    return inserted;
                  });
          LOG.debug(
              "Created {}/{}/{} task instances for workflow {}",
              cnt,
              pendingTasks.size(),
              tasks.size(),
              tasks.get(0).getWorkflowInstanceId());
          return new ArrayList<>(pendingTasks);
        },
        "createTasks",
        "Fail creating {} task instances for workflow: {}",
        tasks.size(),
        tasks.get(0).getWorkflowInstanceId());
  }
@Test
  public void testCreateTaskException() {
    Task task = new Task();
    task.setScheduledTime(1L);
    task.setSeq(1);
    task.setTaskId(TEST_TASK_ID_1);
    task.setTaskDefName(TEST_TASK_DEF_NAME);

    expectedException.expect(NullPointerException.class);
    expectedException.expectMessage("Workflow instance id cannot be null");
    getExecutionDAO().createTasks(Collections.singletonList(task));

    task.setWorkflowInstanceId(TEST_WORKFLOW_ID);
    expectedException.expect(NullPointerException.class);
    expectedException.expectMessage("Task reference name cannot be null");
    getExecutionDAO().createTasks(Collections.singletonList(task));
  }