Handle request cancellation before headers arrive and avoid onErrorDropped logs

Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>

Author: chemicL

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -521,7 +521,10 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 
 					if (transportSession
 						.markInitialized(httpResponse.headers().firstValue("mcp-session-id").orElse(null))) {
-						reconnect(null).contextWrite(deliveredSink.contextView()).subscribe();
+						// Fails only when the transport has been closed in the meantime,
+						// in which case there is no stream left to open.
+						reconnect(null).contextWrite(deliveredSink.contextView()).subscribe(ignored -> {
+						}, t -> logger.debug("Not opening the SSE stream: {}", t.getMessage()));
 					}
 
 					if (statusCode < 200 || statusCode >= 300) {

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseBodyHandlers.java

@@ -17,6 +17,9 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Publisher;
 
@@ -201,16 +204,42 @@ static <T> Flux<T> cancel(Publisher<List<ByteBuffer>> body) {
 	 * Such a body must be subscribed to, or the connection it is read from is never
 	 * released. Should the exchange be cancelled once the response has arrived but before
 	 * its body could be subscribed to, the response is discarded, and its body cancelled.
+	 *
+	 * <p>
+	 * Cancelling the exchange before the response has arrived aborts the request. The
+	 * {@link HttpClient} then fails its future with a {@link CompletionException}
+	 * wrapping a {@link CancellationException}, which {@link Mono#fromFuture} does not
+	 * recognise as the outcome of its own cancellation and reports as a dropped error.
+	 * Only this method can cancel the future, so such a failure is always the expected
+	 * outcome of cancelling, and is ignored.
 	 * @param httpClient the client to send the request with
 	 * @param request the request to send
 	 */
 	static Mono<HttpResponse<Publisher<List<ByteBuffer>>>> sendAsync(HttpClient httpClient, HttpRequest request) {
-		return Mono.fromFuture(() -> httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofPublisher()))
-			.doOnDiscard(HttpResponse.class, response -> {
-				if (response.body() instanceof Publisher<?> body) {
-					cancelBody(body);
+		return Mono.<HttpResponse<Publisher<List<ByteBuffer>>>>create(sink -> {
+			CompletableFuture<HttpResponse<Publisher<List<ByteBuffer>>>> exchange = httpClient.sendAsync(request,
+					HttpResponse.BodyHandlers.ofPublisher());
+			sink.onCancel(() -> exchange.cancel(true));
+			exchange.whenComplete((response, error) -> {
+				if (error == null) {
+					// Once cancelled, the response is discarded rather than emitted.
+					sink.success(response);
+					return;
+				}
+				Throwable cause = error instanceof CompletionException && error.getCause() != null ? error.getCause()
+						: error;
+				if (cause instanceof CancellationException) {
+					sink.success();
+				}
+				else {
+					sink.error(cause);
 				}
 			});
+		}).doOnDiscard(HttpResponse.class, response -> {
+			if (response.body() instanceof Publisher<?> body) {
+				cancelBody(body);
+			}
+		});
 	}
 
 	private static void cancelBody(Publisher<?> body) {

mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseBodyHandlersSendAsyncTests.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.Disposable;
+import reactor.core.publisher.Hooks;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ResponseBodyHandlersSendAsyncTests {
+
+	private final ExecutorService executor = Executors.newCachedThreadPool();
+
+	private final CountDownLatch releaseResponse = new CountDownLatch(1);
+
+	private final List<Throwable> dropped = new CopyOnWriteArrayList<>();
+
+	private HttpServer server;
+
+	@AfterEach
+	void tearDown() {
+		Hooks.resetOnErrorDropped();
+		this.releaseResponse.countDown();
+		if (this.server != null) {
+			this.server.stop(0);
+		}
+		this.executor.shutdownNow();
+	}
+
+	@Test
+	void cancellingBeforeTheResponseArrivesDropsNoError() throws Exception {
+		CountDownLatch requestReceived = new CountDownLatch(1);
+		this.server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+		this.server.setExecutor(this.executor);
+		this.server.createContext("/", exchange -> {
+			// Never responds, so that the exchange is cancelled while awaiting headers.
+			requestReceived.countDown();
+			try {
+				this.releaseResponse.await();
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			exchange.close();
+		});
+		this.server.start();
+		Hooks.onErrorDropped(this.dropped::add);
+
+		HttpRequest request = HttpRequest
+			.newBuilder(URI.create("http://127.0.0.1:" + this.server.getAddress().getPort() + "/"))
+			.build();
+		Disposable exchange = ResponseBodyHandlers.sendAsync(HttpClient.newHttpClient(), request).subscribe();
+		assertThat(requestReceived.await(5, TimeUnit.SECONDS)).isTrue();
+
+		// The HttpClient fails the aborted exchange within cancel() itself.
+		exchange.dispose();
+
+		assertThat(this.dropped).isEmpty();
+	}
+
+}