I got pretty excited when I read this response, but no, it doesn't work. I'm not sure how this would work since annotated tasks return an xcom object.

Can you point me to the documentation on this function? It's possible I'm not using it correctly.

I can do something like this, which works locally, but breaks when deployed:

    res = annotated_task_function(...)
    res.operator.task_id = 'manually assigned task id'

@task.python(task_id="this_is_my_task_name")

def my_func():

...

This still has the problem that, when you call my_func multiple times in the same dag, the resulting tasks will be labelled, my_func, my_func__1, my_func__2, ...

How about the dynamic task mapping that is now available in 2.3?