ValueError: Tried to get output 'train_data_out' from task 'featureengineering', but task 'featureengineering' not found.

[CedrickArmel]

Hello,
I'm a beginer with KFP. Actually i use KFP v2. In oder to test my components and pipelines I'm using kip.local.DockerRunner().
When running my pipeline the two first step complete successfully but when it's to the last one to start I get this; ValueError: Tried to get output 'train_data_out' from task 'featureengineering', but task 'featureengineering' not found.

I need your help to debug and make it work. Thank you.

Here is my code.

"""The container components & the pipeline definitions"""

import os

from dotenv import load_dotenv
from kfp.dsl import (container_component, ContainerSpec, Dataset,
                     importer, Input, Output, pipeline)
from kfp import local
from omegaconf import OmegaConf

local.init(runner=local.DockerRunner())

load_dotenv()
fe_cfg = OmegaConf.load("conf/base/feature_engineering.yaml")
IMAGE = os.getenv("IMAGE")
RAW_SOURCE = os.getenv("RAW_SOURCE")

TRAIN_DATASET = os.path.join(str(RAW_SOURCE), fe_cfg.components.train_dataset)
TEST_DATASET = os.path.join(str(RAW_SOURCE), fe_cfg.components.test_dataset)


@container_component
def featureengineering(train_set: Input[Dataset],
                       test_set: Input[Dataset],
                       train_data_out: Output[Dataset],
                       test_data_out: Output[Dataset]):
    return ContainerSpec(
        image=IMAGE,
        command=["python"],
        args=["src/feature_engineering/components.py",
              "feature_engineering",
              "--train-set", train_set.path,
              "--test-set", test_set.path,
              "--train-data-out", train_data_out.path,
              "--test-data-out", test_data_out.path])


@container_component
def storefeatures(train_features: Input[Dataset],
                  test_features: Input[Dataset]):
    return ContainerSpec(
        image=IMAGE,
        command=["python"],
        args=["src/feature_engineering/components.py",
              "store_features",
              "--train-features", train_features.path,
              "--test-features", test_features.path])


@pipeline
def feature_engineering_pipeline():
    """Feature engineering pipeline.
    """
    train_set_op = importer(artifact_uri=TRAIN_DATASET,
                            artifact_class=Dataset,
                            reimport=False)
    test_set_op = importer(artifact_uri=TEST_DATASET,
                           artifact_class=Dataset,
                           reimport=False)
    feature_engineering_op = featureengineering(
        train_set=train_set_op.output,
        test_set=test_set_op.output)

     # My pipeline fails here even though the outputs are available at the pipeline_root path

    storefeatures(
        train_features=feature_engineering_op.outputs["train_data_out"],
        test_features=feature_engineering_op.outputs["test_data_out"])

if __name__ == "__main__":
    pipeline_task = feature_engineering_pipeline()
    print(pipeline_task)

"""featueengineering logic"""
def featureengineering(train_set: str,
                       test_set: str,
                       train_data_out: str,
                       test_data_out: str) -> None:
    """Performs the feature engineering step of the pipeline.

    Args:
        train_set (str): /path/to/training_set/ or \
            gs://bucket/path/to/training_set/
        test_set (str): /path/to/validation_set/ or \
            gs://bucket/path/to/validation_set/
        train_data_out (str): /path/to/output/the/train/features/dataset
        test_data_out (str): /path/to/output/the/test/features/dataset
    """
    GCSFileSystem(token=APPLICATION_CREDENTIALS)
    config = OmegaConf.load("conf/base/feature_engineering.yaml")
    columns = config.components.columns
    labels = config.components.labels
    target = config.components.target
    train_data = pd.read_pickle(train_set)[0]
    test_data = pd.read_pickle(test_set)[0]
    train_data, test_data = drop_useless(columns,
                                         train_data=train_data,
                                         test_data=test_data)
    train_data = train_data.dropna(axis=0,
                                   subset=columns)
    test_data = test_data.dropna(axis=0,
                                 subset=columns)
    train_data[target] = train_data[target].\
        apply(lambda cell: encode_cell(cell, labels))
    test_data[target] = test_data[target].\
        apply(lambda cell: encode_cell(cell, labels))

    if not os.path.exists(os.path.dirname(train_data_out)):
        os.makedirs(os.path.dirname(train_data_out))
    if not os.path.exists(os.path.dirname(test_data_out)):
        os.makedirs(os.path.dirname(test_data_out))

    with open(train_data_out, "wb") as f:
        train_data.to_parquet(f)
    with open(test_data_out, "wb") as f:
        test_data.to_parquet(f)