code: @Override
      public List<Task> createTasks(List<Task> tasks) {
          validateTasks(tasks);
          String workflowId = tasks.get(0).getWorkflowInstanceId();
          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 -> {
                  task.setScheduledTime(System.currentTimeMillis());
                  session.execute(updateTaskLookupStatement.bind(UUID.fromString(workflowId), UUID.fromString(task.getTaskId())));
              });
  
              // update all the tasks in the workflow using batch
              BatchStatement batchStatement = new BatchStatement();
              tasks.forEach(task -> {
                  String taskPayload = toJson(task);
                  batchStatement.add(insertTaskStatement.bind(UUID.fromString(workflowId), 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, UUID.fromString(workflowId), DEFAULT_SHARD_ID));
              session.execute(batchStatement);
  
              // update the total tasks and partitions for the workflow
              session.execute(updateTotalPartitionsStatement.bind(DEFAULT_TOTAL_PARTITIONS, totalTasks, UUID.fromString(workflowId)));
  
              return tasks;
          } catch (ApplicationException e) {
              throw e;
          } catch (Exception 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 ApplicationException(ApplicationException.Code.BACKEND_ERROR, errorMsg, e);
          }
      }
docs: 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