Triggerer code does not seem to be able to use an Airflow connection & variable

[pankajkoti]

Apache Airflow version

main (development)

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I am building a new deferrable operator in the databricks provider. Although the synchronous version of the operator seems to work fine, when using deferrable=True, it's complaining that the Airflow connection isn't defined. Likely it appears that the Triggerer is not fetching the Airflow connection or is failing to fetch the connection.

What you think should happen instead?

The Triggerer code should be able to use the Airflow connections so that the deferrable operators work fine.

How to reproduce

Run an example DAG with an Airflow operator with deferrable=True that tries to use an Airflow connection

Operating System

Darwin

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

Locally using latest code from main using Breeze

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[pankajkoti]

This is the traceback I see

Trigger failed:
Traceback (most recent call last):

  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 764, in cleanup_finished_triggers
    result = details["task"].result()

  File "/opt/airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py", line 864, in run_trigger
    async for event in trigger.run():

  File "/opt/airflow/providers/databricks/src/airflow/providers/databricks/triggers/databricks.py", line 178, in run
    statement_state = await self.hook.a_get_sql_statement_state(self.statement_id)

  File "/opt/airflow/providers/databricks/src/airflow/providers/databricks/hooks/databricks.py", line 790, in a_get_sql_statement_state
    response = await self._a_do_api_call(get_sql_statement_endpoint)

  File "/opt/airflow/providers/databricks/src/airflow/providers/databricks/hooks/databricks_base.py", line 707, in _a_do_api_call
    url = self._endpoint_url(endpoint)

  File "/opt/airflow/providers/databricks/src/airflow/providers/databricks/hooks/databricks_base.py", line 619, in _endpoint_url
    port = f":{self.databricks_conn.port}" if self.databricks_conn.port else ""

  File "/usr/local/lib/python3.10/functools.py", line 981, in __get__
    val = self.func(instance)

  File "/opt/airflow/providers/databricks/src/airflow/providers/databricks/hooks/databricks_base.py", line 138, in databricks_conn
    return self.get_connection(self.databricks_conn_id)

  File "/opt/airflow/airflow-core/src/airflow/hooks/base.py", line 64, in get_connection
    conn = Connection.get_connection_from_secrets(conn_id)

  File "/opt/airflow/airflow-core/src/airflow/models/connection.py", line 496, in get_connection_from_secrets
    raise AirflowNotFoundException(f"The conn_id `{conn_id}` isn't defined")

airflow.exceptions.AirflowNotFoundException: The conn_id `databricks_default` isn't defined

[kaxil]

It is same for XCom, Variable or anything that requires DB connection in the actual trigger code.

[vatsrahul1001]

@kaxil for XCom, Variable are we tracking on SDK board?

[jroachgolf84]

Just curious, what would a fix for this look like?

[ashb]

@jroachgolf84 Adding something like

airflow/airflow-core/src/airflow/dag_processing/processor.py

Lines 261 to 296 in 6a703dd

	@functools.cached_property
	def client(self) -> Client:
	from airflow.sdk.api.client import Client
	client = Client(base_url=None, token="", dry_run=True, transport=in_process_api_server().transport)
	# Mypy is wrong -- the setter accepts a string on the property setter! `URLType = URL | str`
	client.base_url = "http://in-process.invalid./" # type: ignore[assignment]
	return client
	def _handle_request(self, msg: ToManager, log: FilteringBoundLogger) -> None: # type: ignore[override]
	from airflow.sdk.api.datamodels._generated import ConnectionResponse, VariableResponse
	resp = None
	if isinstance(msg, DagFileParsingResult):
	self.parsing_result = msg
	return
	elif isinstance(msg, GetConnection):
	conn = self.client.connections.get(msg.conn_id)
	if isinstance(conn, ConnectionResponse):
	conn_result = ConnectionResult.from_conn_response(conn)
	resp = conn_result.model_dump_json(exclude_unset=True, by_alias=True).encode()
	else:
	resp = conn.model_dump_json().encode()
	elif isinstance(msg, GetVariable):
	var = self.client.variables.get(msg.key)
	if isinstance(var, VariableResponse):
	var_result = VariableResult.from_variable_response(var)
	resp = var_result.model_dump_json(exclude_unset=True).encode()
	else:
	resp = var.model_dump_json().encode()
	else:
	log.error("Unhandled request", msg=msg)
	return
	if resp:
	self.stdin.write(resp + b"\n")

here

airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py

Lines 299 to 315 in 6a703dd

	def _handle_request(self, msg: ToTriggerSupervisor, log: FilteringBoundLogger) -> None: # type: ignore[override]
	if isinstance(msg, messages.TriggerStateChanges):
	log.debug("State change from async process", state=msg)
	if msg.events:
	self.events.extend(msg.events)
	if msg.failures:
	self.failed_triggers.extend(msg.failures)
	for id in msg.finished or ():
	self.running_triggers.discard(id)
	self.cancelling_triggers.discard(id)
	# Remove logger from the cache, and since structlog doesn't have an explicit close method, we
	# only need to remove the last reference to it to close the open FH
	if factory := self.logger_cache.pop(id, None):
	factory.upload_to_remote()
	return
	raise ValueError(f"Unknown message type {type(msg)}")

[kaxil]

@kaxil for XCom, Variable are we tracking on SDK board?

No but the same code should fix both, the one that is in Ash's example for DAG processor

[pierrejeambrun]

@jroachgolf84 Thanks for the PR. I didn't know you were working on that.

I also have a draft there:
#48239

I think you are missing some pieces (extend the ToTriggerRunner and ToTriggerSupervisor, socket initialization, first message check etc.)