[
https://issues.apache.org/jira/browse/AIRFLOW-270?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Tamas Szuromi updated AIRFLOW-270:
----------------------------------
Description:
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:
```
[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
```
was:
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 the operator.
I got the following error:
```
[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
```
> 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, 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:
> ```
> [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
> ```
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
