@task
def simple_map_task(a: int) -> int:  # noqa: D103
    return a * a


@task
def simple_reduce(b: list[int]) -> int:  # noqa: D103
    return sum(b)


@task
def to_list(size: int) -> list[int]:  # noqa: D103
    return list(range(size))

@workflow
def map_reduce_array_node(size: int) -> int:
    """Simple workflow to illustrate a large fan-out/fan-in using Flyte experimental feature."""
    input_array = to_list(size=size)
    output = array_node_map_task(simple_map_task)(a=input_array)
    return simple_reduce(b=output)

Workflow[flytesnacks:development:fan_out.map_reduce_array_node] failed. RuntimeExecutionError: max number of system retry attempts [31/30] exhausted. Last known status message: worker error(s) encountered: [0]: failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [armada]: [CorruptedPluginState] Failed to unmarshal custom state in Handle, caused by: gob: wrong type (armada.ExecutionPhase) for received field PluginState.Phase
[1]: failed at Node[n1]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [armada]: [CorruptedPluginState] Failed to unmarshal custom state in Handle, caused by: gob: wrong type (armada.ExecutionPhase) for received field PluginState.Phase

func (p Executor) Handle(ctx context.Context, tCtx core.TaskExecutionContext) (core.Transition, error) {
	incomingState := ExecutionState{}
	logger.Infof(ctx, "Handling")

	// We assume here that the first time this function is called, the custom state we get back is whatever we passed in,
	// namely the zero-value of our struct.
	if _, err := tCtx.PluginStateReader().Get(&incomingState); err != nil {
		logger.Errorf(ctx, "Plugin %s failed to unmarshal custom state when handling [%s] [%s]",
			p.id, tCtx.TaskExecutionMetadata().GetTaskExecutionID().GetGeneratedName(), err)
		return core.UnknownTransition, errors.Wrapf(errors.CorruptedPluginState, err,
			"Failed to unmarshal custom state in Handle")
	}

	// Do what needs to be done, and give this function everything it needs to do its job properly
	outgoingState, transformError := HandleExecutionState(ctx, tCtx, incomingState, p.submitClient, p.jobsClient)

	// Return if there was an error
	if transformError != nil {
		return core.UnknownTransition, transformError
	}

	// If no error, then infer the new Phase from the various states
	phaseInfo := MapExecutionStateToPhaseInfo(outgoingState)

	// Hack to ensure the output reader is set. This is similar to what the k8s pod plugin management does
	if phaseInfo.Phase() == core.PhaseSuccess {
		opReader := ioutils.NewRemoteFileOutputReader(ctx, tCtx.DataStore(), tCtx.OutputWriter(), 0)
		err := tCtx.OutputWriter().Put(ctx, opReader)
		if err != nil {
			return core.UnknownTransition, err
		}
	}

	if err := tCtx.PluginStateWriter().Put(pluginStateVersion, outgoingState); err != nil {
		return core.UnknownTransition, err
	}

	return core.DoTransition(phaseInfo), nil
}

type ExecutionState struct {
	Phase ExecutionPhase

	// This will store the job ID from Armada
	JobSetId string `json:"jobSetId,omitempty"`
	JobId    string `json:"jobId,omitempty"`

	ArmadaEndpoint string
    CreationFailureCount int `json:"creation_failure_count,omitempty"`
}

K\u0003\u0001\u0001\u000bPluginState\u0001\ufffd\ufffd\u0000\u0001\u0003\u0001\u0005Phase\u0001\u0006\u0000\u0001\u000eK8sPluginState\u0001\ufffd\ufffd\u0000\u0001\u000fLastEventUpdate\u0001\ufffd\ufffd\u0000\u0000\u0000?\ufffd\ufffd\u0003\u0001\u0001\u000bPluginState\u0001\ufffd\ufffd\u0000\u0001\u0003\u0001\u0005Phase\u0001\u0004\u0000\u0001\u000cPhaseVersion\u0001\u0006\u0000\u0001\u0006Reason\u0001\u000c\u0000\u0000\u0000\u0010\ufffd\ufffd\u0005\u0001\u0001\u0004Time\u0001\ufffd\ufffd\u0000\u0000\u0000\u0005\ufffd\ufffd\u0002\u0000\u0000"

\ufffd\ufffd\u0003\u0001\u0001\u000eExecutionState\u0001\ufffd\ufffd\u0000\u0001\u0005\u0001\u0005Phase\u0001\u0004\u0000\u0001\u0008JobSetId\u0001\u000c\u0000\u0001\u0005JobId\u0001\u000c\u0000\u0001\u000eArmadaEndpoint\u0001\u000c\u0000\u0001\u0014CreationFailureCount\u0001\u0004\u0000\u0000\u0000G\ufffd\ufffd\u0001\u0002\u0001$b9497c10-9e4b-413c-a786-fa6ef6a607be\u0001\u001a01hzna22my5b8hsa6ytbhrz94b\u0000"

CurrentPhase

PreviousPhase

Phase

Phase

{
  "json": {
    "exec_id": "f3a025450ebba49e7a40",
    "ns": "flytesnacks-development",
    "res_ver": "81883862",
    "routine": "worker-69",
    "wf": "flytesnacks:development:fan_out.map_reduce_array_node"
  },
  "level": "error",
  "msg": "Error when trying to reconcile workflow. Error [worker error(s) encountered: [0]: failed at Node[n0]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [armada]: [CorruptedPluginState] Failed to unmarshal custom state in Handle, caused by: gob: type mismatch: no fields matched compiling decoder for ExecutionState\n[1]: failed at Node[n1]. RuntimeExecutionError: failed during plugin execution, caused by: failed to execute handle for plugin [armada]: [CorruptedPluginState] Failed to unmarshal custom state in Handle, caused by: gob: type mismatch: no fields matched compiling decoder for ExecutionState\n]. Error Type[*errors.errorString]",
  "ts": "2024-06-05T23:09:26Z"
}

map_task

// TODO - if we want to support more plugin types we need to figure out the best way to store plugin state
	// currently just mocking based on node phase -> which works for all k8s plugins
	// we can not pre-allocated a bit array because max size is 256B and with 5k fanout node state = 1.28MB
	pluginStateBytes := a.pluginStateBytesStarted

k8s.PluginState

taskExecutionContext.GeneratedName()