amoghrajesh commented on issue #59093: URL: https://github.com/apache/airflow/issues/59093#issuecomment-3617092617
The problem is usage of `XCom.get_value` in context of API server. `XCom` comes from `BaseXCom` class in task sdk and is _supposed_ to be used in that context itself. The difference betwen plugin registered links and links defined on operators is that the plugin links aren't pushed to XCom: `finalize()` only pushes `task.operator_extra_links` (operator-defined): https://github.com/apache/airflow/blob/df94ff60878bab6c905b04d0507d2703803a3dbc/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L1397-L1400 The plugin links call `get_link()` from API server directly via the UI which calls `get_link()`, which uses XCom.get_value() and fails in API server context. I think a plausabile solution for you would be to update the links which are used in / registered via plugins and probably use the `XcomModel` / session to get the right xcom. Example of your plugin but updated: ```python from airflow.models.xcom import XComModel from airflow.utils.session import create_session from airflow.providers.common.compat.sdk import BaseOperator, BaseOperatorLink from airflow.models.taskinstancekey import TaskInstanceKey from airflow.plugins_manager import AirflowPlugin from airflow.providers.standard.operators.python import PythonOperator class RerunTask(BaseOperatorLink): name = "Rerun Task" operators = [PythonOperator] def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str: with create_session() as session: result = session.execute( XComModel.get_many( key="data_doc", run_id=ti_key.run_id, dag_ids=ti_key.dag_id, task_ids=ti_key.task_id, map_indexes=ti_key.map_index, ).with_only_columns(XComModel.value) ).first() if result: value = XComModel.deserialize_value(result) return value or "https://www.astronomer.io/"; class PythonOperatorLink(BaseOperatorLink): name = "GE Result" def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str: with create_session() as session: result = session.execute( XComModel.get_many( key="data_doc", run_id=ti_key.run_id, dag_ids=ti_key.dag_id, task_ids=ti_key.task_id, map_indexes=ti_key.map_index, ).with_only_columns(XComModel.value) ).first() if result: value = XComModel.deserialize_value(result) return value or 'https://google.com' class AirflowExtraLinkPlugin(AirflowPlugin): name = "extra_link_plugin" operator_extra_links = [ RerunTask(), ] global_operator_extra_links = [ PythonOperatorLink(), RerunTask(), ] ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
