fix: Prevent stats from breaking executor initializer

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂

[kacpermuda]

After changes introduced in #63932, the OL executor initializer was failing, first due to the fact that the stats_factory can still expect arg (it was removed, with no warnings or compat), then the Ol conf was used instead of Airflow conf, it could not work.

Fixed the issues, we now fallback to old way in case the new one fails, and also added a catch-all try/except instead of just ImportError - we do not want stat-related error to break entire dag event emission. It's hard to test the whole scheduler flow with the processPoolExecutor and the initializer, but I've added some unit tests for the _executor_initializer itself, that will prevent such issues in the future.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[xBis7]

@kacpermuda As far as I understand, this line

Stats.initialize(factory=stats_utils.get_stats_factory(Stats))

was added in commit #65677, but it hasn't been included to the latest openlineage release. This is the code from the last release and it's not there

https://github.com/apache/airflow/blob/providers-openlineage/2.15.0/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

The only reason, I can think of that you are getting an error, is that you are running in production the current version of the provider code against the latest release of airflow-core which doesn't have the stats changes from this PR

#63932

[kacpermuda]

Correct, perhaps I should have been more clear, but that's exactly what I'm doing in my e2e tests - installing providers from latest main and run it on different airflow versions (including latest release of 3.2, without the core changes from #63932). And yes, the Ol provider was not yet released, so it's good we'll fix that before it is. The #65677 indeed added the stats the the Ol executor path, but the #63932 introduced changes in core that actually broke it. I'm not saying that is 100% bad, we should be able to make changes to core quasi-freely, it's just introducing a requirement, that the Airflow 3.2 with that latest change, will require the newest OL provider, otherwise it will not be emitting dag events. So I guess we just need to make sure that next release of Airflow has constraints for the latest OL provider, with that fix.

[xBis7]

I see you are right. This is what the initialization looks like in the scheduler in 3.2.1

stats_factory = stats_utils.get_stats_factory(Stats)
Stats.initialize(factory=stats_factory)

https://github.com/apache/airflow/blob/3.2.1/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1460-L1461

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

https://github.com/apache/airflow/actions/runs/25169748334/job/73789176552?pr=63932

Sorry for the inconvenience.

[kacpermuda]

I'm wondering why the CI back compat step for 3.2.1 didn't catch it. It was green on my PR.

There were no tests for the executor initializer in OL provider, so that's why it was all green. I've added them in this PR.

Also wondering if this is the only impact of this change (the Ol provider breaking here), not sure how "public" was the get_stats_factory function (and passing the new export_legacy_names arg to initialize, that was not there before too), but if it changes between minor versions of Airflow without any deprecation warnings etc, we might hear about it again if there is any other usage out there 😄

[xBis7]

This was intended to be used in airflow-core and task-sdk.

Just like, settings are initialized once here

https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/settings.py#L799-L828

and then inherited to all other processes, I would expect something similar to happen for stats as well.

I don't know how the providers are started and if they inherit any state. From @mobuchowski commit, I assume that's not the case which would make the initalization needed in all providers and the interface very public.

[mobuchowski]

@xBis7 we're a bit specific use case, since we create processes 🙂