@Override
  public boolean exceedsRateLimitPerFrequency(Task task, TaskDef taskDef) {
    int rateLimit =
        taskDef == null ? task.getRateLimitPerFrequency() : taskDef.getRateLimitPerFrequency();
    if (rateLimit <= 0) {
      return false;
    }
    int bucketSize =
        taskDef == null
            ? task.getRateLimitFrequencyInSeconds()
            : taskDef.getRateLimitFrequencyInSeconds();
    String taskName = task.getTaskDefName();
    try {
      return withRetryableQuery(
          GET_RUNNING_TASK_COUNT_BY_NAME_STATEMENT,
          statement -> {
            statement.setString(1, taskName);
            statement.setLong(2, System.currentTimeMillis() - 1000 * bucketSize);
          },
          result -> {
            if (result.next()) {
              int cnt = result.getInt(COUNT_COLUMN);
              if (cnt > rateLimit) {
                LOG.info(
                    "Got {} running instance for the task name {} in the past {} second exceeding a limit {}",
                    cnt,
                    taskName,
                    bucketSize,
                    rateLimit);
                return true;
              } else {
                LOG.debug(
                    "Got {} running instance for the task name {} in the past {} second within a limit {}",
                    cnt,
                    taskName,
                    bucketSize,
                    rateLimit);
              }
            }
            return false;
          });
    } catch (Exception e) {
      LOG.warn("Failed checking rate limit for task {} due to {}", taskName, e.getMessage());
      return true;
    }
  }
@Test
  public void testExceedsRateLimitWhenNoRateLimitSet() {
    TaskDef taskDef = createTaskDef(0, 0);
    Task task = createRunningTestTask(TEST_TASK_ID_1);
    assertFalse(dao.exceedsRateLimitPerFrequency(task, taskDef));
  }