[AIRFLOW-245] Add access to task instance to executors

[alexandrnikitin]

Dear Airflow Maintainers,

Please accept this PR that addresses the following issues:

• AIRFLOW-245

Testing Done:

Reminders for contributors:

• Your PR's title must reference an issue on Airflow's JIRA. For example, a PR called "[AIRFLOW-1] My Amazing PR" would close JIRA issue Improving the search functionality in the graph view #1. Please open a new issue if required!
• Please squash your commits when possible and follow the 7 rules of good Git commits.

[codecov-io]

Current coverage is 68.03%

Merging #1596 into master will not change coverage

@@             master      #1596   diff @@
==========================================
  Files           116        116          
  Lines          8310       8310          
  Methods           0          0          
  Messages          0          0          
  Branches          0          0          
==========================================
  Hits           5654       5654          
  Misses         2656       2656          
  Partials          0          0          

Powered by Codecov. Last updated by 8aa7160...464fa92

[bolkedebruin]

Please make your commit follow the guidelines (description, commit message etc). In addition please include tests. Thanks!

[alexandrnikitin]

@bolkedebruin Thank you for the reply. I changed the commit. Regarding tests: There's not a single test of any of the executors. I don't know where to start from 😞

[jlowin]

Something important to keep in mind -- the current version of execute_async() just requires the task key (a bunch of strings) and the command that runs the task (another string, something like airflow run task). That's because we make no assumptions about how the execution is implemented, or on what machine/environment, or even when it will be run. Passing a task object to the execute_async() method is contrary to that assumption. In particular, there is no guarantee that task objects are serializable.

The key should give enough information to load the task inside the executor by querying the database for the matching taskinstance -- I think that would be the far preferable way to implement the desired behavior in a custom executor.

[alexandrnikitin]

I close it for now.