Lee-W commented on code in PR #53532:
URL: https://github.com/apache/airflow/pull/53532#discussion_r2217751689
##########
task-sdk/src/airflow/sdk/definitions/mappedoperator.py:
##########
@@ -736,7 +736,19 @@ def unmap(self, resolve: None | Mapping[str, Any]) ->
BaseOperator:
is_setup = kwargs.pop("is_setup", False)
is_teardown = kwargs.pop("is_teardown", False)
on_failure_fail_dagrun = kwargs.pop("on_failure_fail_dagrun",
False)
- kwargs["task_id"] = self.task_id
+
+ # Fix task_id duplication issue: if task_group is present and has
prefix enabled,
+ # strip the prefix from task_id to avoid double prefixing when
BaseOperator.__init__ is called
+ if self.task_group and self.task_group.prefix_group_id:
+ # Strip the task group prefix to avoid double prefixing
+ prefix = f"{self.task_group.group_id}."
+ if self.task_id.startswith(prefix):
+ kwargs["task_id"] = self.task_id[len(prefix) :]
+ else:
+ kwargs["task_id"] = self.task_id
+ else:
+ kwargs["task_id"] = self.task_id
Review Comment:
I just resolved one of the static checks. let's see how the CI works this
time
--
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]
