@Override
    public List<TaskModel> createTasks(List<TaskModel> tasks) {
        validateTasks(tasks);
        String workflowId = tasks.get(0).getWorkflowInstanceId();
        UUID workflowUUID = toUUID(workflowId, "Invalid workflow id");
        try {
            WorkflowMetadata workflowMetadata = getWorkflowMetadata(workflowId);
            int totalTasks = workflowMetadata.getTotalTasks() + tasks.size();
            // TODO: write into multiple shards based on number of tasks

            // update the task_lookup table
            tasks.forEach(
                    task -> {
                        if (task.getScheduledTime() == 0) {
                            task.setScheduledTime(System.currentTimeMillis());
                        }
                        session.execute(
                                updateTaskLookupStatement.bind(
                                        workflowUUID, toUUID(task.getTaskId(), "Invalid task id")));
                    });

            // update all the tasks in the workflow using batch
            BatchStatement batchStatement = new BatchStatement();
            tasks.forEach(
                    task -> {
                        String taskPayload = toJson(task);
                        batchStatement.add(
                                insertTaskStatement.bind(
                                        workflowUUID,
                                        DEFAULT_SHARD_ID,
                                        task.getTaskId(),
                                        taskPayload));
                        recordCassandraDaoRequests(
                                "createTask", task.getTaskType(), task.getWorkflowType());
                        recordCassandraDaoPayloadSize(
                                "createTask",
                                taskPayload.length(),
                                task.getTaskType(),
                                task.getWorkflowType());
                    });
            batchStatement.add(
                    updateTotalTasksStatement.bind(totalTasks, workflowUUID, DEFAULT_SHARD_ID));
            session.execute(batchStatement);

            // update the total tasks and partitions for the workflow
            session.execute(
                    updateTotalPartitionsStatement.bind(
                            DEFAULT_TOTAL_PARTITIONS, totalTasks, workflowUUID));

            return tasks;
        } catch (DriverException e) {
            Monitors.error(CLASS_NAME, "createTasks");
            String errorMsg =
                    String.format(
                            "Error creating %d tasks for workflow: %s", tasks.size(), workflowId);
            LOGGER.error(errorMsg, e);
            throw new TransientException(errorMsg, e);
        }
    }
Inserts tasks into the Cassandra datastore. <b>Note:</b> Creates the task_id to workflow_id
mapping in the task_lookup table first. Once this succeeds, inserts the tasks into the
workflows table. Tasks belonging to the same shard are created using batch statements.

@param tasks tasks to be created
cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java