Fix deferrable execution_timeout handling in AirbyteTriggerSyncOperator

[SameerMesiah97]

Description

This change fixes an issue (as reported in Issue #64048) affecting AirbyteTriggerSyncOperator in deferrable mode that was not fully resolved by PR #64051.

Previously, execution_timeout was passed directly to defer(). The operator now explicitly passes timeout=None to defer() while still preserving execution_deadline handling within the trigger/operator flow.

Rationale

In deferrable mode, Airbyte job cancellation is performed within execute_complete() when the trigger emits a timeout event.

However, framework-level deferred timeout handling bypasses execute_complete() entirely and does not invoke on_kill() in the triggerer process. As a result, Airbyte jobs could continue running after the Airflow task timed out, leading to leaked workloads and excessive resource consumption.

While this weakens the framework-level deferred timeout guarantee, in this case preventing leaked external workloads takes precedence because there is currently no equivalent cleanup path in the triggerer process.

This change keeps timeout handling within the trigger/operator flow so Airbyte job cancellation can occur correctly.

Tests

Added an operator-level regression test verifying that defer() is called with timeout=None while execution_deadline handling remains preserved for Airbyte job timeout cancellation processing.

Backwards Compatibility

This change does not modify public APIs or method signatures.

Related: #64048

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

• Yes (please specify the tool below)

Generated-by: [GPT 5.5] following the guidelines

[henry3260]

Thanks for your PR!

[henry3260]

I think there is an edge case here. If execution_timeout=30s, but the Airbyte job reaches SUCCEEDED after 45s, the trigger can still mark the task as successful because it checks the Airbyte job status before checking execution_deadline:

airflow/providers/airbyte/src/airflow/providers/airbyte/triggers/airbyte.py

Lines 83 to 102 in 16ad479

	if job_run_status == JobStatusEnum.SUCCEEDED:
	yield TriggerEvent(
	{
	"status": "success",
	"message": f"Job run {self.job_id} has completed successfully.",
	"job_id": self.job_id,
	}
	)
	return
	elif job_run_status == JobStatusEnum.CANCELLED:
	yield TriggerEvent(
	{
	"status": "cancelled",
	"message": f"Job run {self.job_id} has been cancelled.",
	"job_id": self.job_id,
	}
	)
	return
	if self.execution_deadline is not None:

After this PR changes defer(timeout=None), the framework-level deferred timeout no longer catches this case. Should the trigger check execution_deadline before accepting SUCCEEDED / CANCELLED, so execution_timeout remains a hard task-level limit?

[SameerMesiah97]

@henry3260

I agree. This is a very valid edge case but this PR was submitted on short notice to fix a resource leak before an RC2 can be cut for the airbyte provider. For the fix in this PR to land in RC2, it has to be merged by 2026-05-26 so the scope is purposely being kept very tight so that even a generalist reviewer can confidently sign off on it. I believe it might be better for me to open another PR to address your comment. If you disagree with my reasoning, you are free to express yourself here.