alibaba/spring-ai-alibaba · error · IllegalArgumentException

Node stream must return Map result using Data.done(),

Error message

Node stream must return Map result using Data.done(),

What it means

In NodeExecutor.processGraphResponseFlux, when a node's stream completes, the final done value must be a Map (state update, via Data.done()), while certain special values (InterruptionMetadata, BaseCheckpointSaver.Tag) are handled explicitly. Anything else triggers IllegalArgumentException 'Node stream must return Map result using Data.done(),' telling the developer the stream's completion contract was violated.

Source

Thrown at spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/executor/NodeExecutor.java:588

			Map<String, Object> partialStateWithoutFlux = partialState.entrySet()
					.stream()
					.filter(e -> !(e.getValue() instanceof Flux) 
							&& !(e.getValue() instanceof GraphFlux)
							&& !(e.getValue() instanceof ParallelGraphFlux))
					.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

			Map<String, Object> updateState = new HashMap<>();
			if (nodeResultValue.isPresent()) {
				Object value = nodeResultValue.get();
				if (value instanceof Map<?, ?>) {
					updateState = (Map<String, Object>) value;
				}
				else if (value instanceof BaseCheckpointSaver.Tag) {
					// When releaseThread=true, completion may return a Tag.
					// Tag is a checkpoint-release receipt, not a state update, so skip merge here.
				}
				else {
					throw new IllegalArgumentException("Node stream must return Map result using Data.done(),");
				}
			}

			Map<String, Object> combinedUpdateState = new HashMap<>(partialStateWithoutFlux);
			combinedUpdateState.putAll(updateState);
			Optional<InterruptionMetadata> interruptAfterMetadata = interruptAfterForStreaming(context, combinedUpdateState);

			context.mergeIntoCurrentState(partialStateWithoutFlux);
			context.mergeIntoCurrentState(updateState);

			try {
				Command nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());
				context.setNextNodeId(nextCommand.gotoNode());

				context.buildNodeOutputAndAddCheckpoint(updateState);

				context.doListeners(NODE_AFTER, null);
				interruptAfterMetadata.ifPresent(context::setReturnFromEmbedWithValue);

View on GitHub (pinned to f82da0b50f)

Solutions

1. Complete the stream with Data.done(Map<String,Object>) — the map is merged into the graph state.
2. If the final value is interruption feedback, return InterruptionMetadata (handled natively); for checkpoint release receipts return BaseCheckpointSaver.Tag.
3. Wrap a POJO result in a state map before calling done: Data.done(Map.of("result", pojo)).

Example fix

// before
return Flux.just(chunk).concatWith(Flux.just(Data.done(myPojo)));
// after
return Flux.just(chunk).concatWith(Flux.just(Data.done(Map.of("result", myPojo))));

if (!(doneValue instanceof Map) && !(doneValue instanceof InterruptionMetadata) && !(doneValue instanceof BaseCheckpointSaver.Tag)) {
    throw new IllegalStateException("Data.done() must carry a Map<String,Object> state update");
}

boolean isValidDoneResult(Object o) {
    return o instanceof Map<?, ?> || o instanceof InterruptionMetadata || o instanceof BaseCheckpointSaver.Tag;
}

try {
    return processGraphResponseFlux(response, context);
} catch (IllegalArgumentException e) {
    if (e.getMessage() != null && e.getMessage().contains("Data.done()")) {
        throw new IllegalStateException("Node stream completed with unsupported done value; wrap result in a Map", e);
    }
    throw e;
}

When it happens

Trigger: A node's embedded Flux completing with Data.done(nonMapValue), or emitting a final element that is neither Map, InterruptionMetadata, nor BaseCheckpointSaver.Tag — e.g. done with a String, POJO, or null wrapper.

Common situations: Custom streaming nodes returning domain objects instead of state maps at completion; misunderstanding that Data.done() must carry a Map<String,Object> state update; refactoring a node from sync (returning POJO) to streaming without wrapping the result in a Map.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• AppTypeNotSupport
• Failed to parse any valid result from streaming response
• Failed to build JSON-RPC streaming payload
• no more elements into iterator
• GraphFlux done result map keys must be String

AI-assisted analysis of alibaba/spring-ai-alibaba@f82da0b50f (2026-09-09). Data as JSON: /api/errors/942e977d8f2b83e7. Report an issue: GitHub.