@Override
  public void addTaskExecutionLogs(List<TaskExecLog> logs) {
    if (logs == null || logs.isEmpty()) {
      return;
    }

    Set<TaskExecLog> taskLogs = new HashSet<>(logs);
    try {
      int cnt =
          withRetryableStatement(
              CREATE_TASK_EXECUTION_LOGS_STATEMENT,
              statement -> {
                for (TaskExecLog taskLog : taskLogs) {
                  statement.setString(1, taskLog.getTaskId());
                  statement.setLong(2, taskLog.getCreatedTime());
                  statement.setString(3, taskLog.getLog());
                  statement.addBatch();
                }
                int[] res = statement.executeBatch();
                return res.length;
              });
      LOG.debug(
          "Created {}/{} task execution logs for a task with id {}",
          cnt,
          logs.size(),
          logs.get(0).getTaskId());
    } catch (Exception e) {
      LOG.warn(
          "Fail creating {} task execution logs for a task with id {} due to {}",
          logs.size(),
          logs.get(0).getTaskId(),
          e.getMessage());
      // ignore error as the execution log is not in the critical path.
    }
  }
@Test
  public void addTaskExecutionLogsTest() {
    List<TaskExecLog> logs = new ArrayList<>();
    logs.add(createLog(TEST_TASK_ID_1, "log1"));
    logs.add(createLog(TEST_TASK_ID_1, "log2"));
    logs.add(createLog(TEST_TASK_ID_1, "log3"));

    dao.addTaskExecutionLogs(logs);

    List<TaskExecLog> indexedLogs =
        tryFindResults(() -> dao.getTaskExecutionLogs(TEST_TASK_ID_1), 3);

    assertEquals(3, indexedLogs.size());

    assertTrue("Not all logs was indexed", indexedLogs.containsAll(logs));
  }