Remove JobRunners back reference from Job by potiuk · Pull Request #30376 · apache/airflow

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk · 2023-03-30T18:26:36Z

This is the final step of decoupling of the job runner from ORM
based BaseJob. After this change, finally we rich the state that
the BaseJob is just a state of the Job being run, but all
the logic is kept in separate "JobRunner" entity which just
keeps the reference to the job. Also it makes sure that
job in each runner is defined as appropriate for each job type:

SchedulerJobRunner, BackfillJobRunner can only use BaseJob
DagProcessorJobRunner, TriggererJobRunner and especially the
LocalTaskJobRunner can keep both BaseJob and it's Pydantic
BaseJobPydantic representation - for AIP-44 usage.

The highlights of this change:

Job does not have job_runner reference any more
Job is a mandatory parameter when creating each JobRunner
run_job method takes as parameter the job (i.e. where the state
of the job is called) and executor_callable - i.e. the method
to run when the job gets executed
heartbeat callback is also passed a generic callable in order
to execute the post-heartbeat operation of each of the job
type
there is no more need to specify job_type when you create
BaseJob, the job gets its type by a simply creating a runner
with the job
DagFileProcessorManager is now merged into DagProcessorJobRunner.
The JobRuner was essentially calling the processormanager that
created processors. In order to make it consistent with other
Runners - all of it has been moved into runner, so that it
starts doing "something" - similar as other runners - rather than
creating processor manager and running it.

This is the final stage of refactoring that was split into
reviewable stages: #30255 -> #30302 -> #30308 -> this PR.

Please check only the last commit.

Closes: #30325

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

potiuk · 2023-03-30T18:40:04Z

I think there are two errors in tests that I will need to fix (let's see for the CI) - but other than that i think this one should be pretty complete and implements what we've been discussing with @mhenc and @vincbeck and @uranusjr before.

potiuk · 2023-03-30T18:41:53Z

Merged this one into DagProcessorJobRunner - this way it is consistent with other Jobs - which "do something" rather than create another object and "run" it.

potiuk · 2023-04-08T15:47:32Z

OK. I got it green finally @uranusjr -> this is the target of the refactor, I would love to merge - if possible - the 4 dependent PRs before we cut-off 2.6.0 branch to make cherry-picking easier for 2.6 branch (cc: @jedcunningham, @ephraimbuddy )

This is the final step of decoupling of the job runner from ORM based BaseJob. After this change, finally we rich the state that the BaseJob is just a state of the Job being run, but all the logic is kept in separate "JobRunner" entity which just keeps the reference to the job. Also it makes sure that job in each runner is defined as appropriate for each job type: * SchedulerJobRunner, BackfillJobRunner can only use BaseJob * DagProcessorJobRunner, TriggererJobRunner and especially the LocalTaskJobRunner can keep both BaseJob and it's Pydantic BaseJobPydantic representation - for AIP-44 usage. The highlights of this change: * Job does not have job_runner reference any more * Job is a mandatory parameter when creating each JobRunner * run_job method takes as parameter the job (i.e. where the state of the job is called) and executor_callable - i.e. the method to run when the job gets executed * heartbeat callback is also passed a generic callable in order to execute the post-heartbeat operation of each of the job type * there is no more need to specify job_type when you create BaseJob, the job gets its type by a simply creating a runner with the job This is the final stage of refactoring that was split into reviewable stages: apache#30255 -> apache#30302 -> apache#30308 -> this PR. Closes: apache#30325

potiuk requested review from XD-DENG, ashb, bbovenzi, dstandish, ephraimbuddy, jedcunningham, kaxil, o-nikolas and ryanahamilton as code owners March 30, 2023 18:26

boring-cyborgBot added area:API Airflow's REST/HTTP API area:CLI area:Scheduler including HA (high availability) scheduler area:webserver Webserver related Issues kind:documentation labels Mar 30, 2023

potiuk force-pushed the remove-runner-back-reference-from-job branch from 4bc3d55 to b10344cCompare March 30, 2023 18:33

potiuk mentioned this pull request Mar 30, 2023
Decouple "job runner" from BaseJob ORM model #30255
Merged

potiuk requested review from Taragolis and uranusjr March 30, 2023 18:40

potiuk commented Mar 30, 2023
View reviewed changes

mhenc approved these changes Mar 31, 2023
View reviewed changes

Comment threadairflow/jobs/scheduler_job_runner.py Outdated

potiuk force-pushed the remove-runner-back-reference-from-job branch 4 times, most recently from 2bf14da to 71920bfCompare April 4, 2023 10:19

potiuk mentioned this pull request Apr 4, 2023
Support serialization to Pydantic models in Internal API #30282
Merged

potiuk force-pushed the remove-runner-back-reference-from-job branch from 71920bf to a858a1eCompare April 6, 2023 06:59

potiuk requested a review from bolkedebruin as a code owner April 6, 2023 06:59

potiuk force-pushed the remove-runner-back-reference-from-job branch from a858a1e to 2514572Compare April 7, 2023 10:34

potiuk requested a review from eladkal as a code owner April 7, 2023 10:34

potiuk added the AIP-44 Airflow Internal API label Apr 8, 2023

potiuk force-pushed the remove-runner-back-reference-from-job branch 2 times, most recently from 88b4146 to 79b62f2Compare April 10, 2023 09:25

potiuk changed the title ~~Finally decouple job_runner from job~~Remove JobRunners back reference from JobApr 10, 2023

potiuk force-pushed the remove-runner-back-reference-from-job branch 2 times, most recently from 68815f8 to 8e2bb8bCompare April 10, 2023 12:47

potiuk mentioned this pull request Apr 10, 2023
Rename JobRunner modules to *_job_runner and base_job* to job #30302
Merged

potiuk force-pushed the remove-runner-back-reference-from-job branch from 8e2bb8b to bdca1efCompare April 10, 2023 12:58

potiuk commented Apr 10, 2023
View reviewed changes

Comment threadairflow/models/__init__.py Outdated

potiuk commented Apr 10, 2023
View reviewed changes

Comment threadairflow/models/__init__.py Outdated

potiuk force-pushed the remove-runner-back-reference-from-job branch 4 times, most recently from 8889b44 to 727f3afCompare April 10, 2023 22:18

potiuk mentioned this pull request Apr 10, 2023
Separate and split run job method into prepare/execute/complete steps #30308
Merged

ephraimbuddy added the changelog:skip Changes that should be skipped from the changelog (CI, tests, etc..) label Apr 11, 2023

potiuk force-pushed the remove-runner-back-reference-from-job branch from 727f3af to 7ee67a6Compare April 11, 2023 10:48

ephraimbuddy approved these changes Apr 11, 2023
View reviewed changes

Comment threadairflow/jobs/job.py Outdated
Comment threadtests/cli/commands/test_jobs_command.py Outdated

potiuk force-pushed the remove-runner-back-reference-from-job branch 4 times, most recently from 1a251d8 to 1ce3d6aCompare April 11, 2023 19:15

potiuk force-pushed the remove-runner-back-reference-from-job branch from 1ce3d6a to 18b6b5bCompare April 11, 2023 21:50

potiuk merged commit 4403419 into apache:mainApr 11, 2023

potiuk deleted the remove-runner-back-reference-from-job branch April 11, 2023 22:58

Jie211 mentioned this pull request Jun 5, 2024
job table missing dag_id and executor_class for BackfillJob type #40055
Open
2 tasks