From 7a5c7d56f13dbd9756df47732ed3d3c156116225 Mon Sep 17 00:00:00 2001 From: Ikhun Um Date: Sun, 4 May 2025 17:45:09 +0900 Subject: [PATCH] Fix a bug where an HTTP/1 request could failed with `ClosedSessionException` when send concurrently. Motivation: `ClosedSessionException` occurred intermittently when sending multiple HTTP/1 requests concurrently. TCP dump showed that the server sent a response, but Nett's `LoggingHandler` did not log a READ event. This may indicate the Channel's internal state might be broken. The root cause appears to be related to the connection acquisition logic. Because when `H1C` was used explicitly, the `ClosedSessionException` wasn't raised. While acquiring a connection, if a pending connection is in progress does not support multiplexing, a new connection is created. Creating a new connection occurs while the acquired connection is still being initialized. https://github.com/netty/netty/blob/4.2/transport-classes-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java#L673-L684 ``` at com.linecorp.armeria.client.HttpChannelPool$ChannelAcquisitionFuture.handlePiggyback(HttpChannelPool.java:852) at com.linecorp.armeria.client.HttpChannelPool$ChannelAcquisitionFuture.lambda$piggyback$0(HttpChannelPool.java:770) at com.linecorp.armeria.client.HttpChannelPool$ChannelAcquisitionFuture.handlePendingPiggybacks(HttpChannelPool.java:906) at com.linecorp.armeria.client.HttpChannelPool$ChannelAcquisitionFuture.complete(HttpChannelPool.java:870) at com.linecorp.armeria.client.HttpChannelPool.notifyConnect(HttpChannelPool.java:514) at com.linecorp.armeria.client.HttpChannelPool.lambda$connect$2(HttpChannelPool.java:374) at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557) at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492) at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636) at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:625) at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:105) at com.linecorp.armeria.client.HttpSessionHandler.tryCompleteSessionPromise(HttpSessionHandler.java:517) at com.linecorp.armeria.client.HttpSessionHandler.userEventTriggered(HttpSessionHandler.java:447) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:398) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:376) at io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:368) at com.linecorp.armeria.client.Http1ResponseDecoder.userEventTriggered(Http1ResponseDecoder.java:328) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:400) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:376) at io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:368) at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireUserEventTriggered(CombinedChannelDuplexHandler.java:430) at io.netty.channel.ChannelInboundHandlerAdapter.userEventTriggered(ChannelInboundHandlerAdapter.java:117) at io.netty.handler.codec.ByteToMessageDecoder.userEventTriggered(ByteToMessageDecoder.java:388) at io.netty.handler.codec.http.HttpObjectDecoder.userEventTriggered(HttpObjectDecoder.java:610) at io.netty.channel.CombinedChannelDuplexHandler.userEventTriggered(CombinedChannelDuplexHandler.java:241) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:398) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:376) at io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:368) at io.netty.handler.logging.LoggingHandler.userEventTriggered(LoggingHandler.java:222) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:398) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:376) at io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:368) at io.netty.channel.DefaultChannelPipeline$HeadContext.userEventTriggered(DefaultChannelPipeline.java:1375) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:396) at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:376) at io.netty.channel.DefaultChannelPipeline.fireUserEventTriggered(DefaultChannelPipeline.java:862) at com.linecorp.armeria.client.HttpClientPipelineConfigurator.finishSuccessfully(HttpClientPipelineConfigurator.java:435) at com.linecorp.armeria.client.HttpClientPipelineConfigurator.lambda$configureAsHttp$1(HttpClientPipelineConfigurator.java:414) at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590) at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:583) at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:559) at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492) at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636) at io.netty.util.concurrent.DefaultPromise.setSuccess0(DefaultPromise.java:625) at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:105) at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:84) at io.netty.channel.kqueue.AbstractKQueueChannel$AbstractKQueueUnsafe.fulfillConnectPromise(AbstractKQueueChannel.java:616) at io.netty.channel.kqueue.AbstractKQueueChannel$AbstractKQueueUnsafe.finishConnect(AbstractKQueueChannel.java:654) at io.netty.channel.kqueue.AbstractKQueueChannel$AbstractKQueueUnsafe.writeReady(AbstractKQueueChannel.java:438) at io.netty.channel.kqueue.KQueueEventLoop.processReady(KQueueEventLoop.java:215) at io.netty.channel.kqueue.KQueueEventLoop.run(KQueueEventLoop.java:296) at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.base/java.lang.Thread.run(Thread.java:1589) ``` It was difficult to figure out the exact reason of the root cause in Netty. However, given the investigation, Netty's internal Channel status seems broken if a connection is created recursively while initializing a new connection. Related issue: #6625 Modifications: - Reschedule the connection attempt to avoid creating a new connection in the middle of initializing the acquired channel. Result: You no longer see an unexpected `ClosedSessionException` when send HTTP/1 concurrently. --- .../armeria/client/HttpChannelPool.java | 6 +- ...Http1ClientClosedSessionExceptionTest.java | 116 ++++++++++++++++++ 2 files changed, 121 insertions(+), 1 deletion(-) create mode 100644 core/src/test/java/com/linecorp/armeria/client/Http1ClientClosedSessionExceptionTest.java diff --git a/core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java b/core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java index d88fa5b3f09..cd67ebc63a3 100644 --- a/core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java +++ b/core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java @@ -847,7 +847,11 @@ private void handlePiggyback(SessionProtocol desiredProtocol, break; case NEW_CONNECTION: timingsBuilder.pendingAcquisitionEnd(); - connect(desiredProtocol, serializationFormat, key, childPromise, timingsBuilder); + // Reschedule the connection attempt to avoid creating a new connection in the middle of + // initializing the acquired channel. + eventLoop.execute(() -> { + connect(desiredProtocol, serializationFormat, key, childPromise, timingsBuilder); + }); break; case PIGGYBACKED_AGAIN: // There's nothing to do because usePendingAcquisition() was called successfully above. diff --git a/core/src/test/java/com/linecorp/armeria/client/Http1ClientClosedSessionExceptionTest.java b/core/src/test/java/com/linecorp/armeria/client/Http1ClientClosedSessionExceptionTest.java new file mode 100644 index 00000000000..2eaa4c330b5 --- /dev/null +++ b/core/src/test/java/com/linecorp/armeria/client/Http1ClientClosedSessionExceptionTest.java @@ -0,0 +1,116 @@ +/* + * Copyright 2025 LINE Corporation + * + * LINE Corporation 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: + * + * https://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 com.linecorp.armeria.client; + +import java.net.InetSocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.linecorp.armeria.common.AggregatedHttpResponse; +import com.linecorp.armeria.common.CommonPools; +import com.linecorp.armeria.common.HttpHeaderNames; +import com.linecorp.armeria.common.SessionProtocol; +import com.linecorp.armeria.common.logging.RequestLog; +import com.linecorp.armeria.internal.testing.NettyServerExtension; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpResponse; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.codec.http.HttpVersion; + +class Http1ClientClosedSessionExceptionTest { + + private static final Logger logger = LoggerFactory.getLogger(Http1ClientClosedSessionExceptionTest.class); + + @RegisterExtension + static NettyServerExtension http1Server = new NettyServerExtension() { + @Override + protected void configure(Channel ch) throws Exception { + ch.pipeline().addLast(new HttpServerCodec()); + ch.pipeline().addLast(new HttpObjectAggregator(1024)); + ch.pipeline().addLast(new ChannelDuplexHandler() { + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + if (msg instanceof FullHttpRequest) { + final HttpResponse response = new DefaultFullHttpResponse( + HttpVersion.HTTP_1_1, HttpResponseStatus.OK); + // Make sure the connection is closed after the response is sent. + response.headers().add(HttpHeaderNames.CONNECTION, "close"); + ctx.writeAndFlush(response) + .addListener(ChannelFutureListener.CLOSE); + } else { + ctx.fireChannelRead(msg); + } + } + }); + } + }; + + @Test + void shouldSendHttp1RequestConcurrently() throws Throwable { + final WebClient client = WebClient.of(SessionProtocol.HTTP, http1Server.endpoint()); + + SessionProtocolNegotiationCache.setUnsupported( + new InetSocketAddress("127.0.0.1", http1Server.address().getPort()), + SessionProtocol.H2C); + final AtomicReference failed = new AtomicReference<>(); + final Runnable loader = () -> { + while (failed.get() == null) { + try (ClientRequestContextCaptor captor = Clients.newContextCaptor()) { + try { + final CompletableFuture future = client.get("/").aggregate(); + final ClientRequestContext ctx = captor.get(); + future.join(); + final RequestLog log = ctx.log().whenComplete().join(); + final Throwable responseCause = log.responseCause(); + if (responseCause != null) { + failed.set(responseCause); + } + } catch (Throwable ex) { + failed.set(new RuntimeException("Failed to receive the response. ctx: " + captor.get(), + ex)); + } + } + } + }; + + for (int i = 0; i < 5; i++) { + CommonPools.blockingTaskExecutor().execute(loader); + } + + for (int i = 0; i < 5; i++) { + Thread.sleep(1000); + if (failed.get() != null) { + throw failed.get(); + } + } + logger.info("All requests completed successfully"); + } +}