This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0d5c39a20e5514183582be3dd70d528fd45d540b Author: James Netherton <[email protected]> AuthorDate: Thu Sep 17 11:54:41 2026 +0100 CAMEL-24864: camel-platform-http-vertx: do not report a lost client connection as a failure pipe() propagates the response write failures that Pump discarded, so a client going away part way through a response is now reported as an error. Treat reads and handler changes after close as expected in AsyncInputStream, and reset a response that has already started rather than failing the routing context, which cannot send an error status once the head is written and only logs the failure a second time. Classify a failure by where it came from rather than by the state of the response. Netty fails the write promise before it reports the connection close to Vert.x, so ctx.response().closed() is often still false when the failure is handled, and a plain client abort was reported through the consumer's ExceptionHandler. AsyncInputStream now records a failure raised while reading the body, so any other pipe failure is a failed response write and is handled as a lost connection. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --- .../platform/http/vertx/AsyncInputStream.java | 80 ++++-- .../http/vertx/ResponseWriteException.java | 28 +++ .../http/vertx/VertxPlatformHttpConsumer.java | 30 ++- .../http/vertx/VertxPlatformHttpSupport.java | 9 +- .../VertxPlatformHttpClientDisconnectTest.java | 280 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 19 ++ 6 files changed, 423 insertions(+), 23 deletions(-) diff --git a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java index 17825ef417d3..501f1fd7596f 100644 --- a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java +++ b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java @@ -55,6 +55,7 @@ public class AsyncInputStream implements ReadStream<Buffer> { private Handler<Buffer> dataHandler; private Handler<Void> endHandler; private Handler<Throwable> exceptionHandler; + private Throwable readFailure; public AsyncInputStream(Vertx vertx, Context context, InputStream inputStream) { this(vertx, context, inputStream, false); @@ -80,7 +81,9 @@ public class AsyncInputStream implements ReadStream<Buffer> { public AsyncInputStream endHandler(Handler<Void> endHandler) { lock.lock(); try { - checkStreamClosed(); + if (closed) { + return this; + } this.endHandler = endHandler; return this; } finally { @@ -92,7 +95,9 @@ public class AsyncInputStream implements ReadStream<Buffer> { public AsyncInputStream exceptionHandler(Handler<Throwable> exceptionHandler) { lock.lock(); try { - checkStreamClosed(); + if (closed) { + return this; + } this.exceptionHandler = exceptionHandler; return this; } finally { @@ -104,9 +109,11 @@ public class AsyncInputStream implements ReadStream<Buffer> { public AsyncInputStream handler(Handler<Buffer> handler) { lock.lock(); try { - checkStreamClosed(); + if (closed) { + return this; + } this.dataHandler = handler; - if (this.dataHandler != null && !this.closed) { + if (this.dataHandler != null) { this.doRead(); } else { queue.clear(); @@ -121,7 +128,9 @@ public class AsyncInputStream implements ReadStream<Buffer> { public AsyncInputStream pause() { lock.lock(); try { - checkStreamClosed(); + if (closed) { + return this; + } queue.pause(); return this; } finally { @@ -133,7 +142,9 @@ public class AsyncInputStream implements ReadStream<Buffer> { public AsyncInputStream resume() { lock.lock(); try { - checkStreamClosed(); + if (closed) { + return this; + } queue.resume(); return this; } finally { @@ -143,18 +154,31 @@ public class AsyncInputStream implements ReadStream<Buffer> { @Override public ReadStream<Buffer> fetch(long amount) { - checkStreamClosed(); - queue.fetch(amount); - return this; + lock.lock(); + try { + if (closed) { + return this; + } + queue.fetch(amount); + return this; + } finally { + lock.unlock(); + } } public void close(Handler<AsyncResult<Void>> handler) { closeInternal(handler); } - private void checkStreamClosed() { - if (this.closed) { - throw new IllegalStateException("Stream closed"); + /** + * The failure raised while reading the underlying {@link InputStream}, or {@code null} if reading has not failed. + */ + Throwable getReadFailure() { + lock.lock(); + try { + return readFailure; + } finally { + lock.unlock(); } } @@ -191,13 +215,18 @@ public class AsyncInputStream implements ReadStream<Buffer> { } private void doRead() { - checkStreamClosed(); doRead(ByteBuffer.allocate(IOHelper.DEFAULT_BUFFER_SIZE)); } private void doRead(ByteBuffer buffer) { lock.lock(); try { + if (closed) { + // Reads can still be scheduled once the stream is closed, either by a queue drain that was already + // pending or by the next iteration of a read chain that is still in flight. The channel is gone, so + // there is nothing left to read. + return; + } if (!readInProgress) { readInProgress = true; Buffer buff = Buffer.buffer(IOHelper.DEFAULT_BUFFER_SIZE); @@ -284,12 +313,27 @@ public class AsyncInputStream implements ReadStream<Buffer> { } private void handleException(Throwable t) { - if (exceptionHandler != null && t instanceof Exception) { - exceptionHandler.handle(t); - } else { - if (LOG.isErrorEnabled()) { - LOG.error("Unhandled error while processing stream", t); + Handler<Throwable> handler; + boolean streamClosed; + lock.lock(); + try { + handler = this.exceptionHandler; + streamClosed = this.closed; + if (!streamClosed) { + readFailure = t; } + } finally { + lock.unlock(); + } + + if (streamClosed) { + // A read that was in flight when the stream was closed fails because the channel is gone. That is + // expected, for instance when the client disconnects part way through the response. + LOG.debug("Stream closed while a read was in progress", t); + } else if (handler != null && t instanceof Exception) { + handler.handle(t); + } else if (LOG.isErrorEnabled()) { + LOG.error("Unhandled error while processing stream", t); } } } diff --git a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/ResponseWriteException.java b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/ResponseWriteException.java new file mode 100644 index 000000000000..b54266620e64 --- /dev/null +++ b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/ResponseWriteException.java @@ -0,0 +1,28 @@ +/* + * 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.platform.http.vertx; + +/** + * Signals that the response body could not be written to the client, as opposed to a failure producing the body. The + * actual failure is the cause. + */ +final class ResponseWriteException extends RuntimeException { + + ResponseWriteException(Throwable cause) { + super(cause.getMessage(), cause, false, false); + } +} diff --git a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpConsumer.java b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpConsumer.java index 9fdaa9a5784e..deedc4835303 100644 --- a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpConsumer.java +++ b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpConsumer.java @@ -84,6 +84,7 @@ public class VertxPlatformHttpConsumer extends DefaultConsumer private static final String PRE_AUTHENTICATED_EXCHANGE = VertxPlatformHttpConsumer.class.getName() + ".preAuthenticatedExchange"; private static final String AUTHORIZATION = "Authorization"; + private static final long HTTP2_INTERNAL_ERROR = 2; private final List<Handler<RoutingContext>> handlers; private final String fileNameExtWhitelist; @@ -373,10 +374,31 @@ public class VertxPlatformHttpConsumer extends DefaultConsumer } private void handleFailure(Exchange exchange, RoutingContext ctx, Throwable failure) { - getExceptionHandler().handleException( - "Failed handling platform-http endpoint " + getEndpoint().getPath(), - failure); - ctx.fail(failure); + // A failed response write is reported before Vert.x has processed the connection close, so the response may + // not be flagged as closed yet + boolean responseWriteFailed = failure instanceof ResponseWriteException; + if (responseWriteFailed) { + failure = failure.getCause(); + } + + if (responseWriteFailed || ctx.response().closed()) { + LOGGER.debug("Client closed the connection of platform-http endpoint {} before the response completed", + getEndpoint().getPath(), failure); + } else { + getExceptionHandler().handleException( + "Failed handling platform-http endpoint " + getEndpoint().getPath(), + failure); + if (!ctx.response().headWritten()) { + ctx.fail(failure); + } else if (!ctx.response().ended() && !ctx.response().closed()) { + // The response has already started, so there is no error status left to send, and failing the + // routing context would only log the failure a second time as an unhandled router exception. + // Reset it directly instead: the client still has to be told that the response it is reading is + // truncated, rather than being left waiting for a body that will never be completed. + ctx.response().reset(HTTP2_INTERNAL_ERROR); + } + } + if (handleWriteResponseError && failure != null) { Exception existing = exchange.getException(); if (existing != null) { diff --git a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java index 6bbc0148dc10..43fd2903069b 100644 --- a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java +++ b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java @@ -247,7 +247,14 @@ public final class VertxPlatformHttpSupport { .to(response) .onComplete(result -> asyncInputStream.close(closeResult -> { if (result.failed()) { - promise.fail(result.cause()); + Throwable cause = result.cause(); + if (cause == asyncInputStream.getReadFailure()) { + promise.fail(cause); + } else { + // The InputStream did not fail, so writing to the response did. That happens when the + // client has gone away, possibly before the response has been flagged as closed. + promise.fail(new ResponseWriteException(cause)); + } } else { promise.complete(); } diff --git a/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpClientDisconnectTest.java b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpClientDisconnectTest.java new file mode 100644 index 000000000000..27dde4a553a1 --- /dev/null +++ b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpClientDisconnectTest.java @@ -0,0 +1,280 @@ +/* + * 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.platform.http.vertx; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import io.vertx.ext.web.RoutingContext; +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.spi.ExceptionHandler; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +class VertxPlatformHttpClientDisconnectTest { + private static final long RESPONSE_SIZE = 256L * 1024 * 1024; + private static final long MAX_PRODUCED_AFTER_ABORT = 16L * 1024 * 1024; + private static final int BYTES_READ_BEFORE_ABORT = 64 * 1024; + private static final int BYTES_BEFORE_STREAM_FAILURE = 64 * 1024; + + @Test + void serverStopsReadingBodyWhenClientAborts() throws Exception { + final CamelContext context = VertxPlatformHttpEngineTest.createCamelContext(); + final CountingInputStream body = new CountingInputStream(RESPONSE_SIZE); + final RecordingExceptionHandler exceptionHandler = new RecordingExceptionHandler(); + context.getRegistry().bind("recordingExceptionHandler", exceptionHandler); + try { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("platform-http:/download?exceptionHandler=#recordingExceptionHandler") + .process(exchange -> { + // stream the body as-is, so that writing the response is what consumes it + exchange.getExchangeExtension().setStreamCacheDisabled(true); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, "application/octet-stream"); + exchange.getMessage().setBody(body); + }); + } + }); + + VertxPlatformHttpEngineTest.startCamelContext(context); + VertxPlatformHttpServer server = context.hasService(VertxPlatformHttpServer.class); + + readSomeThenAbort(server.getPort()); + + assertTrue(body.awaitClose(20, TimeUnit.SECONDS), + "The response body stream was never closed after the client aborted the request"); + + long produced = body.produced(); + assertTrue(produced < MAX_PRODUCED_AFTER_ABORT, + () -> "The server kept reading the response body after the client aborted: produced " + + produced + " bytes of " + RESPONSE_SIZE); + assertTrue(exceptionHandler.handled.isEmpty(), + () -> "The client aborting the request was reported as a failure: " + exceptionHandler.handled); + } finally { + context.stop(); + } + } + + @Test + void failureAfterResponseStartedIsNotReportedAsAnUnhandledRouterFailure() throws Exception { + final CamelContext context = VertxPlatformHttpEngineTest.createCamelContext(); + final List<String> routerErrors = new CopyOnWriteArrayList<>(); + final RecordingExceptionHandler exceptionHandler = new RecordingExceptionHandler(); + context.getRegistry().bind("recordingExceptionHandler", exceptionHandler); + + AbstractAppender appender = new AbstractAppender("CaptureRouterErrors", null, null, true, Property.EMPTY_ARRAY) { + @Override + public void append(LogEvent event) { + if (event.getLevel() == Level.ERROR) { + routerErrors.add(event.getMessage().getFormattedMessage()); + } + } + }; + appender.start(); + Logger routerLogger = (Logger) LogManager.getLogger(RoutingContext.class); + routerLogger.addAppender(appender); + + try { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("platform-http:/failing-download?exceptionHandler=#recordingExceptionHandler") + .process(exchange -> { + exchange.getExchangeExtension().setStreamCacheDisabled(true); + exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, "application/octet-stream"); + exchange.getMessage().setBody(new FailingInputStream(BYTES_BEFORE_STREAM_FAILURE)); + }); + } + }); + + VertxPlatformHttpEngineTest.startCamelContext(context); + VertxPlatformHttpServer server = context.hasService(VertxPlatformHttpServer.class); + + readUntilTheResponseBreaks(server.getPort()); + + assertTrue(routerErrors.isEmpty(), + () -> "A response that failed after it had started was logged as an unhandled router failure: " + + routerErrors); + assertTrue(exceptionHandler.handled.stream().anyMatch(IOException.class::isInstance), + () -> "The body stream failure was not reported to the exception handler: " + + exceptionHandler.handled); + } finally { + routerLogger.removeAppender(appender); + appender.stop(); + context.stop(); + } + } + + private static void readUntilTheResponseBreaks(int port) throws Exception { + try (Socket socket = new Socket()) { + socket.connect(new InetSocketAddress("localhost", port), 5000); + socket.setSoTimeout(10000); + + OutputStream out = socket.getOutputStream(); + out.write("GET /failing-download HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes(StandardCharsets.UTF_8)); + out.flush(); + + InputStream in = socket.getInputStream(); + byte[] buffer = new byte[8192]; + try { + while (in.read(buffer) != -1) { + // drain until the server resets or ends the connection + } + } catch (SocketTimeoutException e) { + fail("The server left the client waiting on a truncated response instead of terminating it"); + } catch (IOException expected) { + // the server resets the stream once the body fails + } + } + } + + private static void readSomeThenAbort(int port) throws Exception { + try (Socket socket = new Socket()) { + socket.connect(new InetSocketAddress("localhost", port), 5000); + socket.setSoTimeout(20000); + + OutputStream out = socket.getOutputStream(); + out.write("GET /download HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes(StandardCharsets.UTF_8)); + out.flush(); + + InputStream in = socket.getInputStream(); + byte[] buffer = new byte[8192]; + int total = 0; + while (total < BYTES_READ_BEFORE_ABORT) { + int read = in.read(buffer); + if (read == -1) { + break; + } + total += read; + } + + socket.setSoLinger(true, 0); + } + } + + private static final class RecordingExceptionHandler implements ExceptionHandler { + + private final List<Throwable> handled = new CopyOnWriteArrayList<>(); + + @Override + public void handleException(Throwable exception) { + handled.add(exception); + } + + @Override + public void handleException(String message, Throwable exception) { + handled.add(exception); + } + + @Override + public void handleException(String message, Exchange exchange, Throwable exception) { + handled.add(exception); + } + } + + private static final class CountingInputStream extends InputStream { + + private final AtomicLong produced = new AtomicLong(); + private final CountDownLatch closed = new CountDownLatch(1); + private final byte[] block = new byte[8192]; + private final long size; + + private CountingInputStream(long size) { + this.size = size; + } + + @Override + public int read() { + byte[] single = new byte[1]; + return read(single, 0, 1) == -1 ? -1 : single[0] & 0xFF; + } + + @Override + public int read(byte[] b, int off, int len) { + long alreadyProduced = produced.get(); + if (alreadyProduced >= size) { + return -1; + } + int count = (int) Math.min(Math.min(len, block.length), size - alreadyProduced); + System.arraycopy(block, 0, b, off, count); + produced.addAndGet(count); + return count; + } + + @Override + public void close() { + closed.countDown(); + } + + private long produced() { + return produced.get(); + } + + private boolean awaitClose(long timeout, TimeUnit unit) throws InterruptedException { + return closed.await(timeout, unit); + } + } + + private static final class FailingInputStream extends InputStream { + + private final byte[] block = new byte[8192]; + private final int bytesBeforeFailure; + private int produced; + + private FailingInputStream(int bytesBeforeFailure) { + this.bytesBeforeFailure = bytesBeforeFailure; + } + + @Override + public int read() throws IOException { + byte[] single = new byte[1]; + return read(single, 0, 1) == -1 ? -1 : single[0] & 0xFF; + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + if (produced >= bytesBeforeFailure) { + throw new IOException("Source failed while the response was being written"); + } + int count = Math.min(Math.min(len, block.length), bytesBeforeFailure - produced); + System.arraycopy(block, 0, b, off, count); + produced += count; + return count; + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc index 4f794e193dd9..5b8b4cd33632 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc @@ -1501,6 +1501,25 @@ cannot serve one origin's response to another. Deployments that relied on credentialed cross-origin requests must list the permitted origins in `camel.server.cors.origins`. +=== camel-platform-http-vertx - response write failures are no longer silent + +A streaming response body (any `InputStream` payload) was written to the client with the deprecated +Vert.x `Pump`, which discards the result of every `write()`. Failures were therefore invisible: the +response write was always reported as successful, even when nothing reached the client. Writing now +uses `pipe()`, which propagates those failures. + +This is a bug fix (CAMEL-24864). A client that disconnects part way through a streaming response is +now noticed, so the body stream is closed and the exchange completes. Previously neither happened, +which leaked an exchange per aborted response — and, for a stream that only ends on an explicit +marker, such as the Server-Sent Events stream produced by `camel-a2a`, also leaked the subscriber +that the stream's `close()` was responsible for removing. + +A lost client connection is logged at DEBUG and is not reported through the consumer's +`ExceptionHandler`, since a client going away is not a route failure. A failure raised while +*producing* the body is still reported as before. Routes using `handleWriteResponseError=true` now +see an exception on the exchange when a response write fails, including when the cause is the client +disconnecting; routes that need to distinguish the two can inspect the exception cause. + === camel-tika The `tika:parse` producer copies the metadata of the parsed document onto the Camel message. Those
