[BUG] VersionConflictEngineException during /_plugins/_flow_framework/workflow (YAML)

[reuschling]

I POST a workflow to a current OpenSearch release server (2.19.1, Linux) with
"/_plugins/_flow_framework/workflow?provision=true&wait_for_completion_timeout=180s"
Header:
Content-Type: application/yaml

The whole workflow is attached. I receive a strange exception:

[2025-03-28T16:26:46,780][INFO ][o.o.r.m.c.i.LocalClusterIndicesClient] [pc-4156] Updating -KZc3ZUBvOneSnnL4Y-7 from .plugins-flow-framework-state
[2025-03-28T16:26:46,781][ERROR][o.o.f.i.FlowFrameworkIndicesHandler] [pc-4156] Failed to update workflow state for -KZc3ZUBvOneSnnL4Y-7 on step create_llm_connector to add resource connector_id -qZc3ZUBvOneSnnL4Y_x
org.opensearch.OpenSearchStatusException: Failed to update data object in index .plugins-flow-framework-state
	at org.opensearch.remote.metadata.client.impl.LocalClusterIndicesClient.lambda$updateDataObjectAsync$7(LocalClusterIndicesClient.java:252) [opensearch-remote-metadata-sdk-2.19.1.0.jar:?]
	at org.opensearch.core.action.ActionListener$1.onFailure(ActionListener.java:90) [opensearch-core-2.19.1.jar:2.19.1]
	at org.opensearch.action.support.TransportAction$1.onFailure(TransportAction.java:124) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.action.update.TransportUpdateAction.lambda$innerExecute$0(TransportUpdateAction.java:207) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.core.action.ActionListener$1.onFailure(ActionListener.java:90) [opensearch-core-2.19.1.jar:2.19.1]
	at org.opensearch.action.support.single.instance.TransportInstanceSingleOperationAction$AsyncSingleAction$1.handleException(TransportInstanceSingleOperationAction.java:245) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.transport.TransportService$ContextRestoreResponseHandler.handleException(TransportService.java:1527) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.transport.TransportService$DirectResponseChannel.processException(TransportService.java:1641) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.transport.TransportService$DirectResponseChannel.sendResponse(TransportService.java:1615) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.transport.TaskTransportChannel.sendResponse(TaskTransportChannel.java:81) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.action.support.single.instance.TransportInstanceSingleOperationAction$ShardTransportHandler$1.onFailure(TransportInstanceSingleOperationAction.java:312) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.onFailure(ThreadContext.java:1002) [opensearch-2.19.1.jar:2.19.1]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:54) [opensearch-2.19.1.jar:2.19.1]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) [?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) [?:?]
	at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]
Caused by: org.opensearch.transport.RemoteTransportException: [pc-4156][192.168.41.146:9300][indices:data/write/update[s]]
Caused by: org.opensearch.index.engine.VersionConflictEngineException: [-KZc3ZUBvOneSnnL4Y-7]: version conflict, required seqNo [601], primary term [51]. current document has seqNo [602] and primary term [51]

No idea what I can do here, no idea for a possible workaround, it looks to me as a bug. I receive the error every time I post this workflow, so it is easy to reproduce

workflow.txt

[reuschling]

I forgot to mention my cluster settings - at least the trusted connector endpoint has to be set in order that the workflow can work:
POST {{openSearchUrl}}/_cluster/settings

{
  "persistent": {
    "plugins": {
      "ml_commons": {
        "only_run_on_ml_node": "false",
        "model_access_control_enabled": "true",
        "allow_registering_model_via_url": "true",
        "native_memory_threshold": "99",
        
        "model_auto_deploy.enable": "true",
        "model_auto_redeploy.enable": "true",
        "model_auto_redeploy.lifetime_retry_times": "100",
        
        "memory_feature_enabled": true,
        "rag_pipeline_feature_enabled": true,
        "agent_framework_enabled": true,
        
        "connector.private_ip_enabled": true,
        "trusted_connector_endpoints_regex": ["^http://serv-.*/v1/chat/completions*$","^https://serv-.*/v1/chat/completions*$","^.*dfki.*$", "^http://pc-.*$", "^https://pc-.*$"]
        
      }
    },
    "plugins.flow_framework.enabled": true
  }
}

[dbwiddis]

Hi @reuschling I think this may be associated with #114 which will be fixed in 2.19.2.

The cause is a race condition where multiple steps are trying to update the workflow state at the same time (and the conflict was not properly handled due to being a nested exception). The log message indicates that both connectors were being updated at the same time, triggering the bug.

As a workaround, you can try re-oragnizing your workflow to prevent these simultaneous updates (until the bug is fixed). Some options:

• make one of the connector steps a previous node input for the other one. The connector ID will be ignored but this will cause them to be executed in series rather than parallel
• add a no-op step with a delay prior to one of the connector creations.

[dbwiddis]

One other possible workaround: just merged #116 so you could checkout and ./gradlew assemble the flow framework plugin 2.19.2-SNAPSHOT and use that zip in your local deployment (you may need to set the version back to 2.19.1 in the build.gradle first)

[reuschling]

Thank you so much @dbwiddis , you made my day. I added a previous_node_inputs entry inside one connector as you mentioned, the error is gone now.