This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 2860757583a5 CAMEL-23957: Synchronize MCP reconnect to protect shared
endpoint tool state (#24587)
2860757583a5 is described below
commit 2860757583a5b6e95d17eaeb2d30adb1ed5fa0a3
Author: matteominin <[email protected]>
AuthorDate: Fri Jul 10 20:48:04 2026 +0200
CAMEL-23957: Synchronize MCP reconnect to protect shared endpoint tool
state (#24587)
* CAMEL-23957: Synchronize MCP reconnect to protect shared endpoint tool
state
OpenAIEndpoint shared the MCP tool state (cachedMcpTools, toolClientMap,
returnDirectTools, toolToServerName) across concurrent exchanges and mutated
it in place during reconnectMcpServer. In a multi-threaded route hitting an
MCP transport error this caused:
- ConcurrentModificationException / partially-rebuilt tool lists for
producers iterating getMcpTools();
- concurrent reconnects for the same server closing each other's fresh
clients and leaking transports/stdio subprocesses;
- stale returnDirectTools and toolToServerName entries kept after
re-listing.
Fix:
- guard reconnect with a per-server ReentrantLock plus a re-check so only
the first thread reconnects and the rest observe the new client;
- publish the tool state as immutable snapshots swapped via volatile
references, so readers stay lock-free and never see a half-rebuilt list;
- prune tools that vanished server-side from all four collections.
Adds OpenAIEndpointMcpReconnectConcurrencyTest covering the three issues
(reader safety, single reconnect, server-side tool pruning).
Co-Authored-By: Claude Opus 4.8
* Consolidate MCP tool state into a single immutable snapshot
Address review feedback on the reconnect concurrency fix:
- Replace the four separately-published volatile collections with one
immutable McpToolState record swapped through a single volatile
reference, eliminating torn reads (a reader could see the new tool
list with the old client map).
- Guard the rebuild-and-publish section with a global lock so two
servers reconnecting concurrently no longer lose each other's update;
the per-server lock still covers the expensive close/connect/listTools.
- Make shutdown safe: doStop swaps to an empty snapshot under the global
lock and sets an mcpStopped flag re-checked inside the locked reconnect
section, so a reconnect racing with stop cannot NPE or leak a transport.
- Re-read the endpoint snapshot per tool call in OpenAIProducer and
OpenAIToolExecutionProducer so the agentic loop observes mid-exchange
reconnects instead of holding a stale (closed) client.
- Name the field/accessors mcpToolState/getMcpToolState/setMcpToolState
to match the type and the mcp* convention.
Co-Authored-By: Claude Opus 4.8
---
.../camel/component/openai/McpToolState.java | 45 ++++
.../camel/component/openai/OpenAIEndpoint.java | 258 ++++++++++++++-------
.../camel/component/openai/OpenAIProducer.java | 14 +-
.../openai/OpenAIToolExecutionProducer.java | 6 +-
.../OpenAIEndpointMcpReconnectConcurrencyTest.java | 241 +++++++++++++++++++
.../openai/OpenAIProducerMcpMockTest.java | 8 +-
.../openai/integration/OpenAIMcpAdvancedIT.java | 6 +-
7 files changed, 473 insertions(+), 105 deletions(-)
diff --git
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/McpToolState.java
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/McpToolState.java
new file mode 100644
index 000000000000..72d2e87d9236
--- /dev/null
+++
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/McpToolState.java
@@ -0,0 +1,45 @@
+/*
+ * 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.camel.component.openai;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.openai.models.chat.completions.ChatCompletionFunctionTool;
+import io.modelcontextprotocol.client.McpSyncClient;
+
+/**
+ * Immutable snapshot of the MCP tool state shared by all concurrent exchanges.
+ */
+record McpToolState(
+ List<ChatCompletionFunctionTool> tools,
+ Map<String, McpSyncClient> toolClientMap,
+ Map<String, String> toolToServerName,
+ Set<String> returnDirectTools) {
+
+ McpToolState {
+ tools = List.copyOf(tools);
+ toolClientMap = Map.copyOf(toolClientMap);
+ toolToServerName = Map.copyOf(toolToServerName);
+ returnDirectTools = Set.copyOf(returnDirectTools);
+ }
+
+ static McpToolState empty() {
+ return new McpToolState(List.of(), Map.of(), Map.of(), Set.of());
+ }
+}
diff --git
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java
index b81d54ccff36..7c7b09b85e4a 100644
---
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java
+++
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java
@@ -26,6 +26,8 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
import javax.net.ssl.KeyManager;
import javax.net.ssl.KeyManagerFactory;
@@ -86,11 +88,12 @@ public class OpenAIEndpoint extends DefaultEndpoint {
private OpenAIClient client;
- private List<ChatCompletionFunctionTool> cachedMcpTools;
- private Map<String, McpSyncClient> toolClientMap;
- private Set<String> returnDirectTools;
+ private final ReentrantLock globalMcpLock = new ReentrantLock();
+ private Map<String, ReentrantLock> mcpClientLocks;
+
+ private volatile McpToolState mcpToolState = McpToolState.empty();
+ private volatile boolean mcpStopped;
private Map<String, Map<String, String>> serverConfigs;
- private Map<String, String> toolToServerName;
public OpenAIEndpoint(String uri, OpenAIComponent component,
OpenAIConfiguration configuration) {
super(uri, component);
@@ -115,28 +118,31 @@ public class OpenAIEndpoint extends DefaultEndpoint {
@Override
protected void doStart() throws Exception {
super.doStart();
+ mcpStopped = false;
client = createClient();
initializeMcpServers();
}
@Override
protected void doStop() throws Exception {
- if (toolClientMap != null) {
- for (McpSyncClient mcpClient : new
HashSet<>(toolClientMap.values())) {
- try {
- mcpClient.closeGracefully();
- } catch (Exception e) {
- LOG.warn("Error closing MCP client: {}", e.getMessage(),
e);
- }
- }
- toolClientMap.clear();
- }
- if (cachedMcpTools != null) {
- cachedMcpTools.clear();
+ Set<McpSyncClient> toClose;
+ globalMcpLock.lock();
+ try {
+ mcpStopped = true;
+ toClose = new HashSet<>(mcpToolState.toolClientMap().values());
+ mcpToolState = McpToolState.empty();
+ } finally {
+ globalMcpLock.unlock();
}
- if (returnDirectTools != null) {
- returnDirectTools.clear();
+
+ for (McpSyncClient mcpClient : toClose) {
+ try {
+ mcpClient.closeGracefully();
+ } catch (Exception e) {
+ LOG.warn("Error closing MCP client: {}", e.getMessage(), e);
+ }
}
+
if (client != null) {
client.close();
client = null;
@@ -152,10 +158,7 @@ public class OpenAIEndpoint extends DefaultEndpoint {
}
LOG.debug("Initializing MCP servers from configuration: {}",
mcpServerConfig.keySet());
- cachedMcpTools = new ArrayList<>();
- toolClientMap = new HashMap<>();
- returnDirectTools = new HashSet<>();
- toolToServerName = new HashMap<>();
+ mcpClientLocks = new HashMap<>();
// Group flat keys by server name: "fs.transportType" -> {"fs":
{"transportType": ...}}
serverConfigs = new HashMap<>();
@@ -168,33 +171,31 @@ public class OpenAIEndpoint extends DefaultEndpoint {
String serverName = key.substring(0, dot);
String property = key.substring(dot + 1);
serverConfigs.computeIfAbsent(serverName, k -> new
HashMap<>()).put(property, String.valueOf(entry.getValue()));
+ mcpClientLocks.putIfAbsent(serverName, new ReentrantLock());
}
+ List<ChatCompletionFunctionTool> tools = new ArrayList<>();
+ Map<String, McpSyncClient> toolClientMap = new HashMap<>();
+ Map<String, String> toolToServerName = new HashMap<>();
+ Set<String> returnDirectTools = new HashSet<>();
+
for (Map.Entry<String, Map<String, String>> entry :
serverConfigs.entrySet()) {
String serverName = entry.getKey();
Map<String, String> props = entry.getValue();
- String transportType = props.get("transportType");
- if (transportType == null) {
+ if (props.get("transportType") == null) {
throw new IllegalArgumentException("mcpServer." + serverName +
".transportType is required");
}
- LOG.debug("Creating MCP transport for server '{}' with type '{}'",
serverName, transportType);
- McpClientTransport transport = createMcpTransport(serverName,
transportType, props);
- Duration timeout =
Duration.ofSeconds(configuration.getMcpTimeout());
- McpSyncClient mcpClient = McpClient.sync(transport)
- .requestTimeout(timeout)
- .initializationTimeout(timeout)
- .build();
- mcpClient.initialize();
+ McpSyncClient mcpClient = createMcpClient(serverName, props);
LOG.debug("MCP server '{}' initialized, listing tools",
serverName);
McpSchema.ListToolsResult toolsResult = mcpClient.listTools();
- List<McpSchema.Tool> tools = toolsResult.tools();
+ List<McpSchema.Tool> serverTools = toolsResult.tools();
- cachedMcpTools.addAll(McpToolConverter.convert(tools));
+ tools.addAll(McpToolConverter.convert(serverTools));
- for (McpSchema.Tool tool : tools) {
+ for (McpSchema.Tool tool : serverTools) {
if (toolClientMap.putIfAbsent(tool.name(), mcpClient) != null)
{
LOG.warn("Duplicate MCP tool name '{}' from server '{}',
using first registered", tool.name(),
serverName);
@@ -202,14 +203,16 @@ public class OpenAIEndpoint extends DefaultEndpoint {
toolToServerName.put(tool.name(), serverName);
}
- if (tool.annotations() != null &&
Boolean.TRUE.equals(tool.annotations().returnDirect())) {
+ if (isReturnDirect(tool)) {
returnDirectTools.add(tool.name());
}
}
- LOG.info("Initialized MCP server '{}' with {} tools: {}",
serverName, tools.size(),
- tools.stream().map(McpSchema.Tool::name).toList());
+ LOG.info("Initialized MCP server '{}' with {} tools: {}",
serverName, serverTools.size(),
+ serverTools.stream().map(McpSchema.Tool::name).toList());
}
+
+ mcpToolState = new McpToolState(tools, toolClientMap,
toolToServerName, returnDirectTools);
}
private McpClientTransport createMcpTransport(String serverName, String
transportType, Map<String, String> props)
@@ -270,6 +273,22 @@ public class OpenAIEndpoint extends DefaultEndpoint {
};
}
+ /**
+ * Creates and initializes an MCP client for the given server.
+ */
+ McpSyncClient createMcpClient(String serverName, Map<String, String>
props) throws Exception {
+ String transportType = props.get("transportType");
+ LOG.debug("Creating MCP transport for server '{}' with type '{}'",
serverName, transportType);
+ McpClientTransport transport = createMcpTransport(serverName,
transportType, props);
+ Duration timeout = Duration.ofSeconds(configuration.getMcpTimeout());
+ McpSyncClient mcpClient = McpClient.sync(transport)
+ .requestTimeout(timeout)
+ .initializationTimeout(timeout)
+ .build();
+ mcpClient.initialize();
+ return mcpClient;
+ }
+
private List<String> parseMcpProtocolVersions() {
String versions = configuration.getMcpProtocolVersions();
if (versions == null || versions.isBlank()) {
@@ -279,29 +298,56 @@ public class OpenAIEndpoint extends DefaultEndpoint {
}
/**
- * Reconnects the MCP server for the given tool name. Closes the old
client, creates a new transport and client,
- * re-lists tools, and updates all internal mappings.
+ * Reconnects the MCP server that owns the given tool. Serializes
concurrent reconnects of the same server via a
+ * per-server lock and skips the reconnect if another thread already
replaced the failed client.
*
- * @param toolName the tool whose server needs reconnecting
- * @return the new McpSyncClient, or null if reconnection failed
+ * @param oldClient the client that failed and should be replaced
+ * @param toolName the tool whose server needs reconnecting
+ * @return the new (or already-reconnected) McpSyncClient, or
null if reconnection failed
*/
- McpSyncClient reconnectMcpServer(String toolName) {
- String serverName = toolToServerName.get(toolName);
+ McpSyncClient reconnectMcpServer(McpSyncClient oldClient, String toolName)
{
+ String serverName = mcpToolState.toolToServerName().get(toolName);
if (serverName == null || serverConfigs == null) {
LOG.warn("Cannot reconnect: no server configuration found for tool
'{}'", toolName);
return null;
}
+ ReentrantLock lock = mcpClientLocks.get(serverName);
+ if (lock == null) {
+ LOG.warn("Cannot reconnect: no lock found for server '{}'",
serverName);
+ return null;
+ }
+
+ lock.lock();
+ try {
+ McpSyncClient currClient =
mcpToolState.toolClientMap().get(toolName);
+ if (currClient != null && currClient != oldClient) {
+ return currClient;
+ }
+
+ LOG.info("Reconnecting MCP server '{}' for tool '{}'", serverName,
toolName);
+ return doReconnectMcpServer(oldClient, serverName);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Performs the actual reconnect work: closes the old client, creates a
new transport and client, re-lists tools,
+ * and republishes the shared tool state. The caller must hold the
per-server lock for {@code serverName}.
+ *
+ * @param oldClient the old McpSyncClient to close
+ * @param serverName the name of the MCP server to reconnect
+ * @return the new McpSyncClient, or null if reconnection failed
+ */
+ private McpSyncClient doReconnectMcpServer(McpSyncClient oldClient, String
serverName) {
Map<String, String> props = serverConfigs.get(serverName);
if (props == null) {
LOG.warn("Cannot reconnect: no configuration found for server
'{}'", serverName);
return null;
}
- LOG.info("Reconnecting MCP server '{}' for tool '{}'", serverName,
toolName);
-
// Close the old client for this server
- McpSyncClient oldClient = toolClientMap.get(toolName);
if (oldClient != null) {
try {
oldClient.closeGracefully();
@@ -311,35 +357,41 @@ public class OpenAIEndpoint extends DefaultEndpoint {
}
try {
- String transportType = props.get("transportType");
- McpClientTransport transport = createMcpTransport(serverName,
transportType, props);
- Duration timeout =
Duration.ofSeconds(configuration.getMcpTimeout());
- McpSyncClient newClient = McpClient.sync(transport)
- .requestTimeout(timeout)
- .initializationTimeout(timeout)
- .build();
- newClient.initialize();
-
- // Re-list tools and update mappings for all tools from this server
- McpSchema.ListToolsResult toolsResult = newClient.listTools();
- List<McpSchema.Tool> tools = toolsResult.tools();
-
- // Remove old mappings for this server
- toolClientMap.entrySet().removeIf(e -> e.getValue() == oldClient);
- cachedMcpTools.removeIf(t -> {
- String name = t.function().name();
- return serverName.equals(toolToServerName.get(name));
- });
-
- // Add new mappings
- cachedMcpTools.addAll(McpToolConverter.convert(tools));
- for (McpSchema.Tool tool : tools) {
- toolClientMap.put(tool.name(), newClient);
- toolToServerName.put(tool.name(), serverName);
-
- if (tool.annotations() != null &&
Boolean.TRUE.equals(tool.annotations().returnDirect())) {
- returnDirectTools.add(tool.name());
+ McpSyncClient newClient = createMcpClient(serverName, props);
+
+ List<McpSchema.Tool> tools = newClient.listTools().tools();
+
+ globalMcpLock.lock();
+ try {
+ if (mcpStopped) {
+ newClient.closeGracefully();
+ return null;
}
+ // Rebuild shared tool state, replacing this server's old
tools with the freshly listed ones
+ Set<String> oldServerTools = toolsForServer(serverName);
+ List<ChatCompletionFunctionTool> newTools = new
ArrayList<>(mcpToolState.tools());
+ Map<String, McpSyncClient> newClientMap = new
HashMap<>(mcpToolState.toolClientMap());
+ Map<String, String> newToolToServer = new
HashMap<>(mcpToolState.toolToServerName());
+ Set<String> newReturnDirect = new
HashSet<>(mcpToolState.returnDirectTools());
+
+ newTools.removeIf(t ->
oldServerTools.contains(t.function().name()));
+ newClientMap.keySet().removeIf(oldServerTools::contains);
+ newToolToServer.keySet().removeIf(oldServerTools::contains);
+ newReturnDirect.removeIf(oldServerTools::contains);
+
+ newTools.addAll(McpToolConverter.convert(tools));
+ for (McpSchema.Tool tool : tools) {
+ newClientMap.put(tool.name(), newClient);
+ newToolToServer.put(tool.name(), serverName);
+ if (isReturnDirect(tool)) {
+ newReturnDirect.add(tool.name());
+ }
+ }
+
+ // Publish immutable snapshots for lock-free readers
+ mcpToolState = new McpToolState(newTools, newClientMap,
newToolToServer, newReturnDirect);
+ } finally {
+ globalMcpLock.unlock();
}
LOG.info("Reconnected MCP server '{}' with {} tools: {}",
serverName, tools.size(),
@@ -351,6 +403,17 @@ public class OpenAIEndpoint extends DefaultEndpoint {
}
}
+ private Set<String> toolsForServer(String serverName) {
+ return mcpToolState.toolToServerName().entrySet().stream()
+ .filter(e -> serverName.equals(e.getValue()))
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toSet());
+ }
+
+ private static boolean isReturnDirect(McpSchema.Tool tool) {
+ return tool.annotations() != null &&
Boolean.TRUE.equals(tool.annotations().returnDirect());
+ }
+
McpSchema.CallToolResult callTool(McpSyncClient mcpClient, String
toolName, Map<String, Object> argsMap) {
McpSchema.CallToolRequest request = new
McpSchema.CallToolRequest(toolName, argsMap, null);
try {
@@ -360,7 +423,7 @@ public class OpenAIEndpoint extends DefaultEndpoint {
throw e;
}
LOG.info("Transport error calling tool '{}', attempting reconnect:
{}", toolName, e.getMessage());
- McpSyncClient newClient = reconnectMcpServer(toolName);
+ McpSyncClient newClient = reconnectMcpServer(mcpClient, toolName);
if (newClient == null) {
throw e;
}
@@ -517,28 +580,47 @@ public class OpenAIEndpoint extends DefaultEndpoint {
return client;
}
- public List<ChatCompletionFunctionTool> getMcpTools() {
- return cachedMcpTools;
+ public void addReturnDirectTool(String toolName) {
+ globalMcpLock.lock();
+ try {
+ Set<String> newReturnDirect = new
HashSet<>(mcpToolState.returnDirectTools());
+ newReturnDirect.add(toolName);
+ mcpToolState = new McpToolState(
+ mcpToolState.tools(), mcpToolState.toolClientMap(),
+ mcpToolState.toolToServerName(), newReturnDirect);
+ } finally {
+ globalMcpLock.unlock();
+ }
}
- public Map<String, McpSyncClient> getToolClientMap() {
- return toolClientMap;
+ public void removeReturnDirectTool(String toolName) {
+ globalMcpLock.lock();
+ try {
+ Set<String> newReturnDirect = new
HashSet<>(mcpToolState.returnDirectTools());
+ newReturnDirect.remove(toolName);
+ mcpToolState = new McpToolState(
+ mcpToolState.tools(), mcpToolState.toolClientMap(),
+ mcpToolState.toolToServerName(), newReturnDirect);
+ } finally {
+ globalMcpLock.unlock();
+ }
}
- public Set<String> getReturnDirectTools() {
- return returnDirectTools;
+ McpToolState getMcpToolState() {
+ return mcpToolState;
}
// Package-private setters for testing
- void setMcpTools(List<ChatCompletionFunctionTool> tools) {
- this.cachedMcpTools = tools;
+
+ void setMcpToolState(McpToolState state) {
+ this.mcpToolState = state;
}
- void setToolClientMap(Map<String, McpSyncClient> map) {
- this.toolClientMap = map;
+ void setServerConfigs(Map<String, Map<String, String>> configs) {
+ this.serverConfigs = configs;
}
- void setReturnDirectTools(Set<String> tools) {
- this.returnDirectTools = tools;
+ void setMcpClientLocks(Map<String, ReentrantLock> locks) {
+ this.mcpClientLocks = locks;
}
}
diff --git
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java
index ab9cf76665e9..93c543f9d4d1 100644
---
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java
+++
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIProducer.java
@@ -26,7 +26,6 @@ import java.util.Base64;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -182,7 +181,7 @@ public class OpenAIProducer extends DefaultAsyncProducer {
applyAdditionalBodyProperties(paramsBuilder, config);
// Add MCP tools to the request if configured
- List<ChatCompletionFunctionTool> mcpTools =
getEndpoint().getMcpTools();
+ List<ChatCompletionFunctionTool> mcpTools =
getEndpoint().getMcpToolState().tools();
boolean hasMcpTools = mcpTools != null && !mcpTools.isEmpty();
if (hasMcpTools) {
for (ChatCompletionFunctionTool tool : mcpTools) {
@@ -415,7 +414,7 @@ public class OpenAIProducer extends DefaultAsyncProducer {
private void processNonStreaming(Exchange exchange,
ChatCompletionCreateParams params, OpenAIConfiguration config)
throws Exception {
- List<ChatCompletionFunctionTool> mcpTools =
getEndpoint().getMcpTools();
+ List<ChatCompletionFunctionTool> mcpTools =
getEndpoint().getMcpToolState().tools();
boolean hasMcpTools = mcpTools != null && !mcpTools.isEmpty();
if (!hasMcpTools || !config.isAutoToolExecution()) {
@@ -452,11 +451,9 @@ public class OpenAIProducer extends DefaultAsyncProducer {
Exchange exchange, ChatCompletionCreateParams params,
OpenAIConfiguration config)
throws Exception {
- Map<String, McpSyncClient> toolClientMap =
getEndpoint().getToolClientMap();
- Set<String> returnDirectTools = getEndpoint().getReturnDirectTools();
int maxIterations = config.getMaxToolIterations();
LOG.debug("Starting agentic loop with maxToolIterations={}, available
tools: {}", maxIterations,
- toolClientMap.keySet());
+ getEndpoint().getMcpToolState().toolClientMap().keySet());
// Rebuild the builder from the immutable params so we can accumulate
messages
ChatCompletionCreateParams.Builder paramsBuilder = params.toBuilder();
@@ -513,7 +510,8 @@ public class OpenAIProducer extends DefaultAsyncProducer {
String toolCallId = toolCall.asFunction().id();
toolCallsLog.add(toolName);
- McpSyncClient mcpClient = toolClientMap.get(toolName);
+ McpToolState mcpToolState = getEndpoint().getMcpToolState();
+ McpSyncClient mcpClient =
mcpToolState.toolClientMap().get(toolName);
if (mcpClient == null) {
throw new IllegalStateException(
"Tool '" + toolName + "' not found in any
configured MCP server");
@@ -532,7 +530,7 @@ public class OpenAIProducer extends DefaultAsyncProducer {
allReturnDirect = false;
} else {
resultContent =
extractTextContent(toolResult.content());
- if (!returnDirectTools.contains(toolName)) {
+ if
(!mcpToolState.returnDirectTools().contains(toolName)) {
allReturnDirect = false;
}
}
diff --git
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java
index 21544f92869e..616ab9733c29 100644
---
a/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java
+++
b/components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolExecutionProducer.java
@@ -124,8 +124,7 @@ public class OpenAIToolExecutionProducer extends
DefaultProducer {
.build()));
// Execute each tool call via MCP and add tool result messages
- Map<String, McpSyncClient> toolClientMap =
getEndpoint().getToolClientMap();
- if (toolClientMap == null || toolClientMap.isEmpty()) {
+ if (getEndpoint().getMcpToolState().toolClientMap().isEmpty()) {
throw new IllegalStateException(
"No MCP tool clients configured on the endpoint. Configure
mcpServer.* parameters.");
}
@@ -136,7 +135,8 @@ public class OpenAIToolExecutionProducer extends
DefaultProducer {
String argsJson = toolCall.asFunction().function().arguments();
String toolCallId = toolCall.asFunction().id();
- McpSyncClient mcpClient = toolClientMap.get(toolName);
+ McpToolState mcpToolState = getEndpoint().getMcpToolState();
+ McpSyncClient mcpClient =
mcpToolState.toolClientMap().get(toolName);
if (mcpClient == null) {
throw new IllegalStateException(
"Tool '" + toolName + "' not found in any configured
MCP server");
diff --git
a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIEndpointMcpReconnectConcurrencyTest.java
b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIEndpointMcpReconnectConcurrencyTest.java
new file mode 100644
index 000000000000..4b444592ba22
--- /dev/null
+++
b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIEndpointMcpReconnectConcurrencyTest.java
@@ -0,0 +1,241 @@
+/*
+ * 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.camel.component.openai;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+
+import com.openai.models.chat.completions.ChatCompletionFunctionTool;
+import io.modelcontextprotocol.client.McpSyncClient;
+import io.modelcontextprotocol.spec.McpSchema;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Concurrency regression tests for CAMEL-23957: MCP reconnect must not
corrupt the shared endpoint tool state.
+ */
+class OpenAIEndpointMcpReconnectConcurrencyTest {
+
+ private static final String SERVER = "test-server";
+ private static final List<String> TOOL_NAMES = List.of("get_weather",
"find_location");
+
+ private final AtomicInteger clientsCreated = new AtomicInteger();
+
+ private volatile List<String> serverTools = TOOL_NAMES;
+
+ /**
+ * Overrides the client-creation seam so reconnects return mocks instead
of opening real transports.
+ */
+ private class TestEndpoint extends OpenAIEndpoint {
+ TestEndpoint(OpenAIComponent component) {
+ super("openai:chat-completion", component, new
OpenAIConfiguration());
+ }
+
+ @Override
+ McpSyncClient createMcpClient(String serverName, Map<String, String>
props) {
+ clientsCreated.incrementAndGet();
+ return newMockClient();
+ }
+ }
+
+ private McpSyncClient newMockClient() {
+ McpSyncClient client = mock(McpSyncClient.class);
+
when(client.listTools()).thenReturn(McpSchema.ListToolsResult.builder(mockTools()).build());
+ return client;
+ }
+
+ private List<McpSchema.Tool> mockTools() {
+ return serverTools.stream()
+ .map(n -> McpSchema.Tool.builder(n, Map.of("type",
"object")).description("mock " + n).build())
+ .toList();
+ }
+
+ private TestEndpoint newEndpoint(McpSyncClient initialClient) {
+ DefaultCamelContext ctx = new DefaultCamelContext();
+ OpenAIComponent component = new OpenAIComponent();
+ component.setCamelContext(ctx);
+ TestEndpoint endpoint = new TestEndpoint(component);
+ endpoint.setCamelContext(ctx);
+
+ Map<String, McpSyncClient> clientMap = new ConcurrentHashMap<>();
+ Map<String, String> toolToServer = new ConcurrentHashMap<>();
+ for (String n : serverTools) {
+ clientMap.put(n, initialClient);
+ toolToServer.put(n, SERVER);
+ }
+
+ endpoint.setMcpToolState(new McpToolState(
+ McpToolConverter.convert(mockTools()),
+ clientMap,
+ toolToServer,
+ ConcurrentHashMap.newKeySet()));
+
+ Map<String, Map<String, String>> serverConfigs = new
ConcurrentHashMap<>();
+ serverConfigs.put(SERVER, Map.of("transportType", "stdio"));
+ endpoint.setServerConfigs(serverConfigs);
+
+ Map<String, ReentrantLock> locks = new ConcurrentHashMap<>();
+ locks.put(SERVER, new ReentrantLock());
+ endpoint.setMcpClientLocks(locks);
+
+ return endpoint;
+ }
+
+ @Test
+ void concurrentReadersAndReconnectsDoNotThrow() throws Exception {
+ TestEndpoint endpoint = newEndpoint(newMockClient());
+
+ int readers = 8;
+ int reconnecters = 4;
+ int iterations = 500;
+ ExecutorService pool = Executors.newFixedThreadPool(readers +
reconnecters);
+ CountDownLatch start = new CountDownLatch(1);
+ ConcurrentLinkedQueue<Throwable> errors = new
ConcurrentLinkedQueue<>();
+
+ for (int i = 0; i < readers; i++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ for (int k = 0; k < iterations; k++) {
+ for (ChatCompletionFunctionTool t :
endpoint.getMcpToolState().tools()) {
+ t.function().name();
+ }
+ for (String name :
endpoint.getMcpToolState().toolClientMap().keySet()) {
+
endpoint.getMcpToolState().toolClientMap().get(name);
+ }
+ endpoint.getMcpToolState().returnDirectTools().size();
+ }
+ } catch (Throwable t) {
+ errors.add(t);
+ }
+ });
+ }
+ for (int i = 0; i < reconnecters; i++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ for (int k = 0; k < iterations; k++) {
+ String tool = TOOL_NAMES.get(k % TOOL_NAMES.size());
+
endpoint.reconnectMcpServer(endpoint.getMcpToolState().toolClientMap().get(tool),
tool);
+ }
+ } catch (Throwable t) {
+ errors.add(t);
+ }
+ });
+ }
+
+ start.countDown();
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(60, TimeUnit.SECONDS), "threads did
not finish in time");
+
+ assertTrue(errors.isEmpty(), "concurrent access threw: " + errors);
+
+ // state consistency checks
+ assertEquals(TOOL_NAMES.size(),
endpoint.getMcpToolState().tools().size());
+ assertEquals(TOOL_NAMES.size(),
endpoint.getMcpToolState().toolClientMap().size());
+ for (String name : TOOL_NAMES) {
+
assertNotNull(endpoint.getMcpToolState().toolClientMap().get(name), "missing
client for " + name);
+ }
+ }
+
+ @Test
+ void sameServerConcurrentReconnectReconnectsOnceAndClosesOldClientOnce()
throws Exception {
+ McpSyncClient initial = newMockClient();
+ TestEndpoint endpoint = newEndpoint(initial);
+ clientsCreated.set(0);
+
+ int threads = 16;
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ CountDownLatch start = new CountDownLatch(1);
+ ConcurrentLinkedQueue<McpSyncClient> results = new
ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Throwable> errors = new
ConcurrentLinkedQueue<>();
+
+ for (int i = 0; i < threads; i++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ results.add(endpoint.reconnectMcpServer(initial,
"get_weather"));
+ } catch (Throwable t) {
+ errors.add(t);
+ }
+ });
+ }
+ start.countDown();
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS), "threads did
not finish in time");
+
+ assertTrue(errors.isEmpty(), "reconnect threw: " + errors);
+
+ // exactly one real reconnect happened and the other threads observed
it and skipped
+ assertEquals(1, clientsCreated.get(), "server should have been
reconnected exactly once");
+
+ // the failed client was closed exactly once
+ verify(initial, times(1)).closeGracefully();
+
+ // every caller observed the same, non-null reconnected client
+ McpSyncClient reconnected =
endpoint.getMcpToolState().toolClientMap().get("get_weather");
+ assertNotNull(reconnected);
+ for (McpSyncClient c : results) {
+ assertEquals(reconnected, c, "all callers should observe the same
reconnected client");
+ }
+ }
+
+ @Test
+ void reconnectPrunesToolsThatVanishedServerSide() {
+ serverTools = List.of("get_weather", "find_location");
+ McpSyncClient initial = newMockClient();
+ TestEndpoint endpoint = newEndpoint(initial);
+ endpoint.addReturnDirectTool("find_location");
+
+ // The server now only advertises 'get_weather', 'find_location'
disappeared
+ serverTools = List.of("get_weather");
+ McpSyncClient reconnected = endpoint.reconnectMcpServer(initial,
"get_weather");
+ assertNotNull(reconnected);
+
+ // 'find_location' must be pruned from every shared collection
+
assertFalse(endpoint.getMcpToolState().toolClientMap().containsKey("find_location"),
"stale entry in toolClientMap");
+
assertFalse(endpoint.getMcpToolState().toolToServerName().containsKey("find_location"),
+ "stale entry in toolToServerName");
+
assertFalse(endpoint.getMcpToolState().returnDirectTools().contains("find_location"),
+ "stale entry in returnDirectTools");
+ assertFalse(endpoint.getMcpToolState().tools().stream().anyMatch(t ->
t.function().name().equals("find_location")),
+ "stale entry in cachedMcpTools");
+
+ // 'get_weather' is still present and consistent across the collections
+
assertTrue(endpoint.getMcpToolState().toolClientMap().containsKey("get_weather"));
+ assertEquals(SERVER,
endpoint.getMcpToolState().toolToServerName().get("get_weather"));
+ assertEquals(1, endpoint.getMcpToolState().tools().size());
+ assertEquals("get_weather",
endpoint.getMcpToolState().tools().get(0).function().name());
+ }
+}
diff --git
a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIProducerMcpMockTest.java
b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIProducerMcpMockTest.java
index ecbd39dd1c5b..d329bd6835ae 100644
---
a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIProducerMcpMockTest.java
+++
b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIProducerMcpMockTest.java
@@ -129,9 +129,11 @@ public class OpenAIProducerMcpMockTest extends
CamelTestSupport {
.build())
.toList();
- endpoint.setMcpTools(McpToolConverter.convert(mcpTools));
- endpoint.setToolClientMap(toolClients);
- endpoint.setReturnDirectTools(returnDirectToolNames);
+ endpoint.setMcpToolState(new McpToolState(
+ McpToolConverter.convert(mcpTools),
+ toolClients,
+ Map.of(),
+ returnDirectToolNames));
}
@Test
diff --git
a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpAdvancedIT.java
b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpAdvancedIT.java
index aeaeb75bdfeb..9e80955a6925 100644
---
a/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpAdvancedIT.java
+++
b/components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/integration/OpenAIMcpAdvancedIT.java
@@ -171,9 +171,9 @@ public class OpenAIMcpAdvancedIT extends OpenAITestSupport {
*/
@Test
void testReturnDirect() throws InterruptedException {
- // Inject "echo" as a returnDirect tool via the public getter (returns
mutable HashSet)
+ // Mark "echo" as a returnDirect tool at runtime
OpenAIEndpoint endpoint = context.getEndpoint(returnDirectEndpointUri,
OpenAIEndpoint.class);
- endpoint.getReturnDirectTools().add("echo");
+ endpoint.addReturnDirectTool("echo");
try {
MockEndpoint mockEndpoint =
getMockEndpoint("mock:return-direct-result");
@@ -199,7 +199,7 @@ public class OpenAIMcpAdvancedIT extends OpenAITestSupport {
"Body should contain the raw tool result but was: " +
body);
} finally {
// Cleanup: remove to avoid affecting other tests
- endpoint.getReturnDirectTools().remove("echo");
+ endpoint.removeReturnDirectTool("echo");
}
}
