Croway commented on code in PR #24587:
URL: https://github.com/apache/camel/pull/24587#discussion_r3559450514
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java:
##########
@@ -311,37 +351,37 @@ McpSyncClient reconnectMcpServer(String toolName) {
}
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();
+ McpSyncClient newClient = createMcpClient(serverName, props);
- // 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));
- });
+ List<McpSchema.Tool> tools = newClient.listTools().tools();
- // Add new mappings
- cachedMcpTools.addAll(McpToolConverter.convert(tools));
- for (McpSchema.Tool tool : tools) {
- toolClientMap.put(tool.name(), newClient);
- toolToServerName.put(tool.name(), serverName);
+ // 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<>(cachedMcpTools);
Review Comment:
**Lost update across servers**: this read-copy-write of all four global
collections runs under only the *per-server* lock. With two MCP servers
configured, servers A and B can reconnect concurrently under different locks,
both snapshot the same base state, and the last publish silently erases the
other's update — leaving that server's tools mapped to its already-closed old
client. Suggest one short global lock (or `synchronized` block) around this
rebuild-and-publish section, keeping the per-server lock for the expensive
close/connect/`listTools()` part.
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java:
##########
@@ -311,37 +351,37 @@ McpSyncClient reconnectMcpServer(String toolName) {
}
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();
+ McpSyncClient newClient = createMcpClient(serverName, props);
- // 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));
- });
+ List<McpSchema.Tool> tools = newClient.listTools().tools();
- // Add new mappings
- cachedMcpTools.addAll(McpToolConverter.convert(tools));
- for (McpSchema.Tool tool : tools) {
- toolClientMap.put(tool.name(), newClient);
- toolToServerName.put(tool.name(), serverName);
+ // 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<>(cachedMcpTools);
+ Map<String, McpSyncClient> newClientMap = new
HashMap<>(toolClientMap);
+ Map<String, String> newToolToServer = new
HashMap<>(toolToServerName);
+ Set<String> newReturnDirect = new HashSet<>(returnDirectTools);
- if (tool.annotations() != null &&
Boolean.TRUE.equals(tool.annotations().returnDirect())) {
- returnDirectTools.add(tool.name());
+ 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
+ cachedMcpTools = List.copyOf(newTools);
Review Comment:
**Torn reads**: these four volatile writes are not atomic as a group — a
reader can observe the new `cachedMcpTools` with the old `toolClientMap` and
hit `IllegalStateException("Tool ... not found")` in `OpenAIProducer` for a
tool the model was legitimately offered. Consolidating the four collections
into one immutable holder swapped via a single volatile reference eliminates
this (and the lost-update issue above) in one refactor.
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java:
##########
@@ -129,14 +132,13 @@ protected void doStop() throws Exception {
LOG.warn("Error closing MCP client: {}", e.getMessage(),
e);
}
}
- toolClientMap.clear();
- }
- if (cachedMcpTools != null) {
- cachedMcpTools.clear();
- }
- if (returnDirectTools != null) {
- returnDirectTools.clear();
+ toolClientMap = null;
}
+
+ cachedMcpTools = null;
+ returnDirectTools = null;
+ toolToServerName = null;
Review Comment:
**Stop/reconnect race**: nulling `toolToServerName` means a reconnect racing
with endpoint stop now NPEs at the first line of `reconnectMcpServer` (the old
code degraded to the "cannot reconnect" warning). A reconnect already past the
null checks can also publish fresh state and a new initialized client *after*
`doStop` closed everything — a leaked transport. Consider acquiring the
per-server locks here, or checking `isStopping()/isStopped()` inside the locked
section. Minor: `mcpClientLocks` and `serverConfigs` are left non-null while
everything else is nulled.
##########
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.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+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);
+
+ endpoint.setMcpTools(new
ArrayList<>(McpToolConverter.convert(mockTools())));
+
+ 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.setToolClientMap(clientMap);
+ endpoint.setToolToServerName(toolToServer);
+ endpoint.setReturnDirectTools(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 {
Review Comment:
Coverage gap: all tests use a single server, so the cross-server lost-update
race (see comment in `doReconnectMcpServer`) isn't exercised. A two-server
variant of this test — reconnecting both servers concurrently and asserting
each server's tools still map to a live client afterwards — would catch it.
Nit: `assertEquals(reconnected, c)` on mocks relies on identity equality;
`assertSame` states the intent directly.
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java:
##########
@@ -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 Map<String, ReentrantLock> mcpClientLocks;
Review Comment:
Minor: `mcpClientLocks` (and `serverConfigs`) are written in `doStart` and
read from exchange threads but are not `volatile`, while the other four fields
now are. Camel's service lifecycle provides the happens-before in practice, but
since this PR is explicitly about visibility, making these `volatile` too keeps
the reasoning uniform.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
