[BUG]: map_task with timeout on tasks fails with min_success_ratio set

[ljstrnadiii]

Describe the bug

A failing task due to timeout should get filtered by map_task's min_success_ratio functionality. Note: timeout seems to be ignored locally and this only fails after registering.

The issue I see in the flyteui: [1/1] currentAttempt done. Last Error: USER::task execution timeout [10s] expired

An mre/snippet:

import time
from flytekit import task, workflow, map_task

@task(timeout=10)
def unpredictable_task(t: int):
    time.sleep(t)

@workflow
def unpredictable_workflow():
    result = map_task(unpredictable_task, min_success_ratio=0.0)(t=[1,2,3,11,12,13])

Expected behavior

I would expect map_task to succeed and for the result to be of length 3 corresponding to the successful tasks.

Additional context to reproduce

No response

Screenshots

No response

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes

[welcome]

Thank you for opening your first issue here! 🛠

[rnatour]

I'm noticing a related bug: it appears that the timeout applies to the entire map_task node, rather than individual tasks. I would expect that the timeout should apply to each task individually.

Going off the parent example:

import time
from flytekit import task, workflow, map_task

@task(timeout=10)
def unpredictable_task(t: int):
    time.sleep(t)

@workflow
def unpredictable_workflow():
    result = map_task(unpredictable_task, concurrency=1)(t=[1,2,3,4,5])

This causes the map_task to fail, since the entire set of tasks takes longer than 10s when they are run sequentially.