You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Cory Taylor (JIRA)" <ji...@apache.org> on 2016/08/03 18:52:20 UTC
[jira] [Comment Edited] (AIRFLOW-270) Scheduler bug with manual run
[ https://issues.apache.org/jira/browse/AIRFLOW-270?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15406366#comment-15406366 ]
Cory Taylor edited comment on AIRFLOW-270 at 8/3/16 6:51 PM:
-------------------------------------------------------------
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}
Here is the celery executor command log:
{code}
Traceback (most recent call last):
File "/usr/lib/python2.7/site-packages/celery/app/trace.py", line 240, in trace_task
R = retval = fun(*args, **kwargs)
File "/usr/lib/python2.7/site-packages/celery/app/trace.py", line 438, in __protected_call__
return self.run(*args, **kwargs)
File "/usr/lib/python2.7/site-packages/airflow/executors/celery_executor.py", line 45, in execute_command
raise AirflowException('Celery command failed')
AirflowException: Celery command failed
{code}
was (Author: coryt):
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)