You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "MadKingZK (via GitHub)" <gi...@apache.org> on 2023/05/09 06:47:48 UTC

[GitHub] [airflow] MadKingZK opened a new issue, #31143: max_active_tis_per_dagrun不生效，报错：dependency 'Task Concurrency' FAILED: The max task concurrency per run has been reached

MadKingZK opened a new issue, #31143:
URL: https://github.com/apache/airflow/issues/31143

   ### Apache Airflow version
   
   2.6.0
   
   ### What happened
   
   当我在task里设置max_active_tis_per_dagrun=1后，并发运行dag时，per_dagrun并没有限制到每个dagrun，而是在dag级别限制并发。
   queued状态的task日志显示：`dependency 'Task Concurrency' FAILED: The max task concurrency per run has been reached.`
   
   
   ### What you think should happen instead
   
   查找到日志产生的函数是：airflow/ti_deps/deps/task_concurrency_dep.py中TaskConcurrencyDep._get_dep_statuses
   [https://github.com/apache/airflow/blob/b8f73768ec13f8d4cc1605cca3fa93be6caac473/airflow/models/taskinstance.py#L2551](url)
   代码中调用的ti.get_num_running_task_instances(session, same_dagrun=True)这个方法中
   ```python
          num_running_task_instances_query = session.query(func.count()).filter(
               TaskInstance.dag_id == self.dag_id,
               TaskInstance.task_id == self.task_id,
               TaskInstance.state == State.RUNNING,
           ) 
           if same_dagrun:
               num_running_task_instances_query.filter(TaskInstance.run_id == self.run_id)
   ```
   
   代码`num_running_task_instances_query.filter(TaskInstance.run_id == self.run_id)`在执行sql时，并未真正将`TaskInstance.run_id == self.run_id`写入到sql中
   
   filter：sqlalchemy/orm/query.py的Query.filter方法
   ```python
   @_generative
       @_assertions(_no_statement_condition, _no_limit_offset)
       def filter(self, *criterion):
   ```
   这个filter不支持链式调用，因为它每次都会返回一个新的object
   
   
   
   
   ### How to reproduce
   
   [https://github.com/apache/airflow/blob/b8f73768ec13f8d4cc1605cca3fa93be6caac473/airflow/models/taskinstance.py#L2551](url)
   代码：
   ```python
   if same_dagrun:
       num_running_task_instances_query.filter(TaskInstance.run_id == self.run_id)
   ```
   
   应该改为：
   ```python
   if same_dagrun:
       num_running_task_instances_query = num_running_task_instances_query.filter(TaskInstance.run_id == self.run_id)
   ```
   
   ### Operating System
   
   centos 6.5
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Virtualenv installation
   
   ### Deployment details
   
   _No response_
   
   ### Anything else
   
   _No response_
   
   ### Are you willing to submit PR?
   
   - [ ] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] potiuk commented on issue #31143: max_active_tis_per_dagrun不生效，报错：dependency 'Task Concurrency' FAILED: The max task concurrency per run has been reached

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on issue #31143:
URL: https://github.com/apache/airflow/issues/31143#issuecomment-1539992688

   I am sorry, but this is an english-spoken forum. If you want to raise an issue, you have to adapt and raise it in English.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] potiuk closed issue #31143: max_active_tis_per_dagrun不生效，报错：dependency 'Task Concurrency' FAILED: The max task concurrency per run has been reached

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed issue #31143: max_active_tis_per_dagrun不生效，报错：dependency 'Task Concurrency' FAILED: The max task concurrency per run has been reached
URL: https://github.com/apache/airflow/issues/31143


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] boring-cyborg[bot] commented on issue #31143: max_active_tis_per_dagrun不生效，报错：dependency 'Task Concurrency' FAILED: The max task concurrency per run has been reached

Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #31143:
URL: https://github.com/apache/airflow/issues/31143#issuecomment-1539558803

   Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org