airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Cory Taylor (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (AIRFLOW-270) Scheduler bug with manual run
Date Wed, 03 Aug 2016 18:32:20 GMT

    [ https://issues.apache.org/jira/browse/AIRFLOW-270?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15406366#comment-15406366
] 

Cory Taylor commented on AIRFLOW-270:
-------------------------------------

Also seeing a similar error.
Airflow v1.7.1.3 with CeleryExecutor

{code}
[2016-08-03 18:23:48,316] {models.py:305} INFO - Finding 'running' jobs without a recent heartbeat
[2016-08-03 18:23:48,317] {models.py:311} INFO - Failing jobs without heartbeat after 2016-08-03
18:21:33.317277
[2016-08-03 18:23:48,337] {jobs.py:758} ERROR - [Errno 32] Broken pipe
Traceback (most recent call last):
  File "/usr/lib/python2.7/site-packages/airflow/jobs.py", line 755, in _execute
    executor.heartbeat()
  File "/usr/lib/python2.7/site-packages/airflow/executors/base_executor.py", line 107, in
heartbeat
    self.sync()
  File "/usr/lib/python2.7/site-packages/airflow/executors/celery_executor.py", line 74, in
sync
    state = async.state
  File "/usr/lib/python2.7/site-packages/celery/result.py", line 394, in state
    return self._get_task_meta()['status']
  File "/usr/lib/python2.7/site-packages/celery/result.py", line 339, in _get_task_meta
    return self._maybe_set_cache(self.backend.get_task_meta(self.id))
  File "/usr/lib/python2.7/site-packages/celery/backends/amqp.py", line 163, in get_task_meta
    binding.declare()
  File "/usr/lib/python2.7/site-packages/kombu/entity.py", line 521, in declare
    self.exchange.declare(nowait)
  File "/usr/lib/python2.7/site-packages/kombu/entity.py", line 174, in declare
    nowait=nowait, passive=passive,
  File "/usr/lib/python2.7/site-packages/amqp/channel.py", line 615, in exchange_declare
    self._send_method((40, 10), args)
  File "/usr/lib/python2.7/site-packages/amqp/abstract_channel.py", line 56, in _send_method
    self.channel_id, method_sig, args, content,
  File "/usr/lib/python2.7/site-packages/amqp/method_framing.py", line 221, in write_method
    write_frame(1, channel, payload)
  File "/usr/lib/python2.7/site-packages/amqp/transport.py", line 182, in write_frame
    frame_type, channel, size, payload, 0xce,
  File "/usr/lib64/python2.7/socket.py", line 224, in meth
    return getattr(self._sock,name)(*args)
error: [Errno 32] Broken pipe
[2016-08-03 18:23:48,342] {jobs.py:759} ERROR - Tachycardia!
{code}

> Scheduler bug with manual run
> -----------------------------
>
>                 Key: AIRFLOW-270
>                 URL: https://issues.apache.org/jira/browse/AIRFLOW-270
>             Project: Apache Airflow
>          Issue Type: Bug
>          Components: scheduler
>    Affects Versions: Airflow 1.7.1.2, Airflow 1.7.1.3
>         Environment: Airflow with rabbitmq, flower and celery
>            Reporter: Tamas Szuromi
>
> If I hit run on operator I got error from the scheduler and it stops scheduling other
tasks.
> This issue only occurs when I manually run an operator.
> I got the following error:
> {code:xml}
> [2016-06-22 12:58:11,407] {jobs.py:498} INFO - Getting list of tasks to skip for active
runs.
> [2016-06-22 12:58:11,407] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances,
minus 0 skippable ones
> [2016-06-22 12:58:11,499] {jobs.py:498} INFO - Getting list of tasks to skip for active
runs.
> [2016-06-22 12:58:11,499] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances,
minus 0 skippable ones
> [2016-06-22 12:58:11,539] {jobs.py:741} INFO - Done queuing tasks, calling the executor's
heartbeat
> [2016-06-22 12:58:11,539] {jobs.py:744} INFO - Loop took: 0.495144 seconds
> [2016-06-22 12:58:11,542] {models.py:305} INFO - Finding 'running' jobs without a recent
heartbeat
> [2016-06-22 12:58:11,543] {models.py:311} INFO - Failing jobs without heartbeat after
2016-06-22 12:55:56.543141
> [2016-06-22 12:58:11,551] {jobs.py:758} ERROR - [Errno 32] Broken pipe
> Traceback (most recent call last):
>   File "/usr/lib/python2.7/dist-packages/airflow/jobs.py", line 755, in _execute
>     executor.heartbeat()
>   File "/usr/lib/python2.7/dist-packages/airflow/executors/base_executor.py", line 107,
in heartbeat
>     self.sync()
>   File "/usr/lib/python2.7/dist-packages/airflow/executors/celery_executor.py", line
74, in sync
>     state = async.state
>   File "/usr/local/lib/python2.7/dist-packages/celery/result.py", line 394, in state
>     return self._get_task_meta()['status']
>   File "/usr/local/lib/python2.7/dist-packages/celery/result.py", line 339, in _get_task_meta
>     return self._maybe_set_cache(self.backend.get_task_meta(self.id))
>   File "/usr/local/lib/python2.7/dist-packages/celery/backends/amqp.py", line 163, in
get_task_meta
>     binding.declare()
>   File "/usr/local/lib/python2.7/dist-packages/kombu/entity.py", line 521, in declare
>     self.exchange.declare(nowait)
>   File "/usr/local/lib/python2.7/dist-packages/kombu/entity.py", line 174, in declare
>     nowait=nowait, passive=passive,
>   File "/usr/local/lib/python2.7/dist-packages/amqp/channel.py", line 615, in exchange_declare
>     self._send_method((40, 10), args)
>   File "/usr/local/lib/python2.7/dist-packages/amqp/abstract_channel.py", line 56, in
_send_method
>     self.channel_id, method_sig, args, content,
>   File "/usr/local/lib/python2.7/dist-packages/amqp/method_framing.py", line 221, in
write_method
>     write_frame(1, channel, payload)
>   File "/usr/local/lib/python2.7/dist-packages/amqp/transport.py", line 182, in write_frame
>     frame_type, channel, size, payload, 0xce,
>   File "/usr/lib/python2.7/socket.py", line 224, in meth
>     return getattr(self._sock,name)(*args)
> error: [Errno 32] Broken pipe
> [2016-06-22 12:58:11,554] {jobs.py:759} ERROR - Tachycardia!
> [2016-06-22 12:58:11,557] {jobs.py:574} INFO - Prioritizing 0 queued jobs
> [2016-06-22 12:58:11,564] {jobs.py:726} INFO - Starting 2 scheduler jobs
> [2016-06-22 12:58:11,641] {jobs.py:741} INFO - Done queuing tasks, calling the executor's
heartbeat
> [2016-06-22 12:58:11,641] {jobs.py:744} INFO - Loop took: 0.086888 seconds
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message