Per Task Retry-Policy

[javrasya]

Description

Defining retry-policy per task as it is defined in #1073

Motivation and Context

It may be required to have different retry-policy for different tasks according to their priority and resource usage. Assume that, a task is retrieving data from Hive which responses in long time and other one is retrieving data from RDBM which responses in short time. I may want Hive task not to retry many times and not to drain all my resource again an again. But, retrying RDBM task many times may not be problem for me. Assume that, I have some network issue for a while and I should be able to give its retry-count more than Hive task. So I would need to define retry-policy at task level.

This fixes the issue defined in #1073

Have you tested this? If so, how?

• Some unit tests are added about the feature.
• Run available tests(Not all of them, I don't know, some of them maybe includes hadoop ones, I couldn't make passed. ) via tox and travis.
• PyCharm Debugging.

What is added?

• The feature
• Unit tests
• Documentation
• An example

[Tarrasch]

Actually we shouldn't have this information here as it's already somewhere else in the documentation.

[javrasya]

I think we should include a part about what is in retry policy and what they can use per task，what do you think about it?

[Tarrasch]

Can you also add a test case with with dynamic dependencies? It should be super easy, and it immediately shows if it's necessary to thread along the deps_retry_policy_dicts var or not.

edit: Oh, I see that you actually have no tests that sets up a task, sets the configuration and then checks that it behaves as expected. You should and this for sure. You basically just need to copy paste your included example but make it into a testcase. And while at it make one version with run()time dependencies.

[javrasya]

@Tarrasch, I added multiple test cases in PerTaskRetryPolicyBehaviorTest in worker_test. You can see deps_retry_policy is important by removing the part below and run PerTaskRetryPolicyBehaviorTest again.

scheduler.py -> add_task()

retry_policy=self._generate_retry_policy(deps_retry_policy_dicts.get(dep))

This is because of luigi task adding order. Luigi adds a dependency task indirectly with its parent task first. So we need to send deps_retry_policy of its parent task unless luigi adds dependency task directly instead of indirectly with its parent task.

[Tarrasch]

You don't need to be so explicit and say record_task_history=False. But it's ok. :)

[Tarrasch]

@javrasya, I still think deps_retry_policy can be removed. To make it clearer how, I created javrasya#1 :)

[Tarrasch]

And this is now finally in!! :)