👋 Hello, team! Recently I've deployed Flyte to AKS and it requires a valid

AZURE_STORAGE_CONNECTION_STRING

environment variable to work, otherwise it will throw this error

flytekit.exceptions.system.FlyteDownloadDataException: SYSTEM:DownloadDataError: error=Failed to get data from <abfs://data-pipeline-alpha/flytesnacks/development/BLYABWOHQLKRP6B5JW2Y7XMYCQ======/fast31e63d3e22ecaf0c9d9eea81c81f9d59.tar.gz> to ./ (recursive=False).

I made it work by defining a pod template

apiVersion: v1
kind: PodTemplate
metadata:
  name: default-template
template:
  spec:
    tolerations:
      - key: "exclusive"
        operator: "Equal"
        value: "only"
        effect: "NoSchedule"
    nodeSelector:
      exclusive: flyte
    containers:
      - name: default
        image: python:3.10-slim
        env:
          - name: AZURE_STORAGE_CONNECTION_STRING
            valueFrom:
              secretKeyRef:
                name: flyte-pod-secret
                key: AZURE_STORAGE_CONNECTION_STRING
    hostNetwork: false
---
apiVersion: v1
kind: Secret
metadata:
  name: flyte-pod-secret
type: Opaque
data:
  AZURE_STORAGE_CONNECTION_STRING: __AZURE_STORAGE_CONNECTION_STRING__ # this will be substitute in CI/CD

and deploy it to the domain

kubectl apply -f template.yaml -n flytesnacks-development

It works with simple workflow:

@task(
    pod_template_name="default-template"
)
def hello_name(name: str) -> str:
    return f"Hello, {name}!"

@workflow
def say_hello(name: str = 'nguyen'):
    say = hello_name(name)

if __name__ == "__main__":
    print(f"Running wf() {say_hello()}")

But when I use map task, it doesn't create the pod with the template above, which makes

hello_name

fail (

get_names

is fine though)

@task(
    pod_template_name="default-template" <============ good
)
def get_names(name: str) -> List[str]:
    results = [name + str(i) for i in range(5)]
    return results

@task(
    pod_template_name="default-template"  <=========== isn't applied to pod
)
def hello_name(name: str) -> str:   
    return f"Hello, {name}!"

@workflow
def say_hello(name: str = 'nguyen'):
    names = get_names(name=name)
    hello_name_map = functools.partial(hello_name)
    map_task(
        hello_name_map,
        metadata=TaskMetadata(cache=True, cache_version="0.1", retries=5),
        concurrency=2,
    )(name=names)

if __name__ == "__main__":
    print(f"Running wf() {say_hello()}")

Any advice?

where is this metadata set?

@freezing-airport-6809 Flytekit Version: 1.14.3