[runtime][java] Make embedded Python interpreters thread-confined

docs/content/docs/development/workflow_agent.md

@@ -657,6 +657,20 @@ To use async execution on JDK 21+, user should append jvm option `--add-exports=
 {{< /tab >}}
 {{< /tabs >}}
 
+{{< hint warning >}}
+Async actions in either language may call cross-language resources. Within one async execution,
+keep the cross-language call chain to one reverse callback. Only the following call shapes are
+supported:
+
+- Python action → Java resource → Python callback
+- Java action → Python resource → Java callback
+
+For example, a Python action may use a Java `ChatModelSetup` backed by a Python
+`ChatModelConnection`. Multiple or deeper cross-language callbacks, such as
+Python→Java→Python→Java or Java→Python→Java→Python, are not recommended and may deadlock
+when an outer call occupies an async or interpreter worker while waiting for another worker.
+{{< /hint >}}
+
 ### Cross-language Actions
 
 An action declared in one language can dispatch its body to the other language by setting a `target` on the decorator/annotation. The decorated function or annotated method then acts as a stub — it should raise so direct calls outside the framework fail loud.

e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/ConcurrentChatModelCrossLanguageAgent.java

@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.resource.test;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.EventType;
+import org.apache.flink.agents.api.InputEvent;
+import org.apache.flink.agents.api.OutputEvent;
+import org.apache.flink.agents.api.agents.Agent;
+import org.apache.flink.agents.api.annotation.Action;
+import org.apache.flink.agents.api.annotation.ChatModelConnection;
+import org.apache.flink.agents.api.annotation.ChatModelSetup;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.chat.model.BaseChatModelConnection;
+import org.apache.flink.agents.api.context.RunnerContext;
+import org.apache.flink.agents.api.event.ChatRequestEvent;
+import org.apache.flink.agents.api.event.ChatResponseEvent;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceName;
+import org.apache.flink.agents.api.tools.Tool;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/** Java agent exercising concurrent Java-to-Python-to-Java chat calls. */
+public class ConcurrentChatModelCrossLanguageAgent extends Agent {
+
+    /** Java connection that only returns after two chat requests overlap. */
+    public static class OverlappingJavaChatModelConnection extends BaseChatModelConnection {
+        private final CountDownLatch concurrentCalls = new CountDownLatch(2);
+
+        public OverlappingJavaChatModelConnection(
+                ResourceDescriptor descriptor, ResourceContext resourceContext) {
+            super(descriptor, resourceContext);
+        }
+
+        @Override
+        public ChatMessage chat(
+                List<ChatMessage> messages, List<Tool> tools, Map<String, Object> modelParams) {
+            concurrentCalls.countDown();
+            try {
+                if (!concurrentCalls.await(30, TimeUnit.SECONDS)) {
+                    throw new IllegalStateException(
+                            "Timed out waiting for concurrent cross-language chat request.");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IllegalStateException(
+                        "Interrupted while waiting for concurrent cross-language chat request.", e);
+            }
+
+            ChatMessage request = messages.get(messages.size() - 1);
+            return new ChatMessage(
+                    MessageRole.ASSISTANT, "java-connection:" + request.getContent());
+        }
+    }
+
+    @ChatModelConnection
+    public static ResourceDescriptor overlappingJavaConnection() {
+        return ResourceDescriptor.Builder.newBuilder(
+                        OverlappingJavaChatModelConnection.class.getName())
+                .build();
+    }
+
+    @ChatModelSetup
+    public static ResourceDescriptor pythonChatModel() {
+        return ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.PYTHON_WRAPPER_SETUP)
+                .addInitialArgument("pythonClazz", ResourceName.ChatModel.Python.OLLAMA_SETUP)
+                .addInitialArgument("connection", "overlappingJavaConnection")
+                .addInitialArgument("model", "mock-model")
+                .addInitialArgument("extract_reasoning", false)
+                .build();
+    }
+
+    @Action(EventType.InputEvent)
+    public static void requestChat(Event event, RunnerContext ctx) {
+        String input = String.valueOf(InputEvent.fromEvent(event).getInput());
+        ctx.sendEvent(
+                new ChatRequestEvent(
+                        "pythonChatModel", List.of(new ChatMessage(MessageRole.USER, input))));
+    }
+
+    @Action(EventType.ChatResponseEvent)
+    public static void emitResponse(Event event, RunnerContext ctx) {
+        ChatResponseEvent response = ChatResponseEvent.fromEvent(event);
+        ctx.sendEvent(new OutputEvent(response.getResponse().getContent()));
+    }
+}

e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/ConcurrentChatModelCrossLanguageTest.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.resource.test;
+
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+import org.apache.flink.agents.api.agents.AgentExecutionOptions;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** E2E coverage for concurrent calls through a Python setup and Java connection. */
+public class ConcurrentChatModelCrossLanguageTest {
+
+    @Test
+    public void testConcurrentPythonSetupWithJavaConnection() throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setParallelism(1);
+
+        DataStream<String> inputStream = env.fromData("first-request", "second-request");
+        AgentsExecutionEnvironment agentsEnv =
+                AgentsExecutionEnvironment.getExecutionEnvironment(env);
+        agentsEnv.getConfig().set(AgentExecutionOptions.NUM_ASYNC_THREADS, 2);
+        agentsEnv.getConfig().set(AgentExecutionOptions.CHAT_ASYNC, true);
+
+        DataStream<Object> outputStream =
+                agentsEnv
+                        .fromDataStream(inputStream, (KeySelector<String, String>) value -> value)
+                        .apply(new ConcurrentChatModelCrossLanguageAgent())
+                        .toDataStream();
+
+        CloseableIterator<Object> results = outputStream.collectAsync();
+        agentsEnv.execute();
+
+        List<String> responses = new ArrayList<>();
+        while (results.hasNext()) {
+            responses.add(String.valueOf(results.next()));
+        }
+
+        assertThat(responses).hasSize(2);
+        assertThat(new HashSet<>(responses))
+                .isEqualTo(
+                        Set.of("java-connection:first-request", "java-connection:second-request"));
+    }
+}

e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/Mem0LongTermMemoryTest.java

@@ -27,7 +27,6 @@
 import org.apache.flink.util.CloseableIterator;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Assumptions;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
@@ -61,7 +60,6 @@
  *       MILVUS_URI} env var
  * </ul>
  */
-@Disabled("Disabled until #1087 is resolved")
 public class Mem0LongTermMemoryTest {
 
     private final boolean embeddingReady;

python/flink_agents/e2e_tests/e2e_tests_resource_cross_language/concurrent_chat_model_cross_language_agent.py

@@ -0,0 +1,109 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+import threading
+from typing import Any, List, Sequence
+
+from pydantic import PrivateAttr
+from typing_extensions import override
+
+from flink_agents.api.agents.agent import Agent
+from flink_agents.api.agents.types import OutputSchema
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_models.chat_model import BaseChatModelConnection
+from flink_agents.api.decorators import action, chat_model_connection, chat_model_setup
+from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
+from flink_agents.api.events.event import Event, InputEvent, OutputEvent
+from flink_agents.api.events.event_type import EventType
+from flink_agents.api.resource import ResourceDescriptor, ResourceName
+from flink_agents.api.runner_context import RunnerContext
+from flink_agents.api.tools.tool import Tool
+
+
+class OverlappingPythonChatModelConnection(BaseChatModelConnection):
+    """Python connection that only returns after two chat requests overlap."""
+
+    _concurrent_calls: threading.Barrier = PrivateAttr(
+        default_factory=lambda: threading.Barrier(2)
+    )
+
+    @override
+    def chat(
+        self,
+        messages: Sequence[ChatMessage],
+        tools: List[Tool] | None = None,
+        output_schema: OutputSchema | None = None,
+        **kwargs: Any,
+    ) -> ChatMessage:
+        """Echo the request after observing another in-flight call."""
+        self._reject_unsupported_output_schema(output_schema)
+        try:
+            self._concurrent_calls.wait(timeout=30)
+        except threading.BrokenBarrierError as error:
+            message = "Timed out waiting for concurrent cross-language chat request."
+            raise RuntimeError(message) from error
+
+        return ChatMessage(
+            role=MessageRole.ASSISTANT,
+            content=f"python-connection:{messages[-1].content}",
+        )
+
+
+class ConcurrentChatModelCrossLanguageAgent(Agent):
+    """Python agent exercising concurrent Python-to-Java-to-Python chat calls."""
+
+    @chat_model_connection
+    @staticmethod
+    def overlapping_python_connection() -> ResourceDescriptor:
+        """Declare the Python connection used behind the Java setup."""
+        return ResourceDescriptor(
+            clazz=(
+                f"{OverlappingPythonChatModelConnection.__module__}."
+                f"{OverlappingPythonChatModelConnection.__name__}"
+            )
+        )
+
+    @chat_model_setup
+    @staticmethod
+    def java_chat_model() -> ResourceDescriptor:
+        """Declare a Java setup backed by the Python connection."""
+        return ResourceDescriptor(
+            clazz=ResourceName.ChatModel.JAVA_WRAPPER_SETUP,
+            java_clazz=ResourceName.ChatModel.Java.OLLAMA_SETUP,
+            connection="overlapping_python_connection",
+            model="mock-model",
+            extract_reasoning=False,
+        )
+
+    @action(EventType.InputEvent)
+    @staticmethod
+    def request_chat(event: Event, ctx: RunnerContext) -> None:
+        """Send one chat request per input key."""
+        input_value = str(InputEvent.from_event(event).input)
+        ctx.send_event(
+            ChatRequestEvent(
+                model="java_chat_model",
+                messages=[ChatMessage(role=MessageRole.USER, content=input_value)],
+            )
+        )
+
+    @action(EventType.ChatResponseEvent)
+    @staticmethod
+    def emit_response(event: Event, ctx: RunnerContext) -> None:
+        """Emit the cross-language chat response."""
+        response = ChatResponseEvent.from_event(event).response
+        ctx.send_event(OutputEvent(output=response.content))