diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/AbstractNettyConnectionClient.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/AbstractNettyConnectionClient.java index fc6a1a0630a6..cb96e348d7d7 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/AbstractNettyConnectionClient.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/AbstractNettyConnectionClient.java @@ -209,7 +209,8 @@ public void onConnected(Object channel) { return; } - // Close the existing channel before setting a new channel + // Close the existing channel before setting a new channel. + // With the closeFuture CAS fix, closing the old channel won't wipe the new one. io.netty.channel.Channel current = getNettyChannel(); if (current != null) { current.close(); @@ -229,10 +230,16 @@ public void onConnected(Object channel) { connectedPromise.trySuccess(null); if (logger.isDebugEnabled()) { - logger.debug("Connection:{} connected", this); + logger.debug("Connection:{} connected, graceful migration complete", this); } } + /** + * Fallback onGoaway: CAS-null the channel and close it. + * In the graceful migration flow (see {@link NettyConnectionHandler#onGoAway}), + * this is only called when doConnect() fails after GOAWAY, as a last resort + * to let the connectivity-scheduler handle recovery. + */ @Override public void onGoaway(Object channel) { if (!(channel instanceof io.netty.channel.Channel)) { @@ -247,7 +254,7 @@ public void onGoaway(Object channel) { } NettyChannel.removeChannelIfDisconnected(nettyChannel); if (logger.isDebugEnabled()) { - logger.debug("Connection:{} goaway", this); + logger.debug("Connection:{} goaway fallback: channel nulled, awaiting scheduler recovery", this); } } } @@ -269,6 +276,14 @@ protected void clearNettyChannel() { channelRef.set(null); } + /** + * CAS-clear the channel reference: only clears if the current reference equals the expected channel. + * Used by closeFuture listeners to prevent wiping a newly-swapped channel during graceful migration. + */ + protected boolean compareAndClearNettyChannel(io.netty.channel.Channel expected) { + return channelRef.compareAndSet(expected, null); + } + @Override @SuppressWarnings("unchecked") public T getChannel(Boolean generalizable) { @@ -316,6 +331,24 @@ public static AbstractConnectionClient getConnectionClientFromChannel(io.netty.c return channel.attr(CONNECTION).get(); } + /** + * Package-private accessor for {@link NettyConnectionHandler} to schedule + * graceful-migration tasks on the connectivity executor (instead of the + * Netty I/O EventLoop, which would block the I/O thread during doConnect). + */ + java.util.concurrent.ScheduledExecutorService getConnectivityExecutor() { + return connectivityExecutor; + } + + /** + * Package-private accessor for the configured reconnect interval, used by + * {@link NettyConnectionHandler} when scheduling the retry that follows a + * failed graceful migration attempt. + */ + long getReconnectDuration() { + return reconnectDuration; + } + public ChannelFuture write(Object request) throws RemotingException { if (!isAvailable()) { throw new RemotingException( diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannel.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannel.java index b9eda700f7ae..47c3dfce4e21 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannel.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyChannel.java @@ -150,6 +150,23 @@ static void removeChannel(Channel ch) { } } + /** + * Look up the existing NettyChannel from cache without creating a new one. + * Used by handlers (e.g. {@link NettyConnectionHandler#channelInactive}) that + * only need to read state from an already-registered channel and must not + * allocate transient unregistered NettyChannel instances when the underlying + * Netty channel is no longer active. + * + * @param ch netty channel + * @return cached NettyChannel or null if absent + */ + static NettyChannel getChannelIfPresent(Channel ch) { + if (ch == null) { + return null; + } + return CHANNEL_MAP.get(ch); + } + @Override public InetSocketAddress getLocalAddress() { return AddressUtils.getLocalAddress(this); diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java index 8fcec6ddc1df..a9ee087446d7 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionClient.java @@ -122,8 +122,10 @@ protected void initChannel(SocketChannel ch) { new Http2ClientSettingsHandler(connectionPrefaceReceivedPromiseRef)); } - // set null but do not close this client, it will be reconnecting in the future - ch.closeFuture().addListener(channelFuture -> clearNettyChannel()); + // Use compareAndSet to prevent wiping a newly-swapped channel during graceful migration. + // When onConnected() sets the new channel and then the old channel closes, + // the old channel's closeFuture must only clear its own reference, not the new one. + ch.closeFuture().addListener(channelFuture -> compareAndClearNettyChannel(ch)); // TODO support Socks5 // set channel initialized promise to success if necessary. diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionHandler.java b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionHandler.java index 2e17e754881e..11a5d42f9dcf 100644 --- a/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionHandler.java +++ b/dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyConnectionHandler.java @@ -18,9 +18,9 @@ import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; import org.apache.dubbo.common.logger.LoggerFactory; -import org.apache.dubbo.remoting.Constants; import org.apache.dubbo.remoting.api.connection.ConnectionHandler; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import io.netty.channel.Channel; @@ -30,8 +30,22 @@ import io.netty.util.Attribute; import io.netty.util.AttributeKey; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_FAILED_RECONNECT; import static org.apache.dubbo.common.constants.LoggerCodeConstants.TRANSPORT_UNEXPECTED_EXCEPTION; +/** + * Connection lifecycle handler for Netty-based Dubbo connections. + * + *

Implements graceful HTTP/2 GOAWAY migration: when a GOAWAY frame is received + * (e.g., from Envoy/Higress max_requests_per_connection), the old channel is kept + * alive for in-flight requests while a new connection is established in the background. + * Once the new connection is ready, {@link AbstractNettyConnectionClient#onConnected} atomically + * swaps the channel reference and closes the old one - eliminating any unavailability window. + * + *

Fallback: if the new connection fails (e.g., mTLS handshake timeout), the handler + * calls {@link AbstractNettyConnectionClient#onGoaway} to null the channel and let the + * connectivity-scheduler handle recovery (original behavior). + */ @Sharable public class NettyConnectionHandler extends ChannelInboundHandlerAdapter implements ConnectionHandler { @@ -39,6 +53,14 @@ public class NettyConnectionHandler extends ChannelInboundHandlerAdapter impleme LoggerFactory.getErrorTypeAwareLogger(NettyConnectionHandler.class); private static final AttributeKey GO_AWAY_KEY = AttributeKey.valueOf("dubbo_channel_goaway"); + + /** + * Delay before initiating new connection after GOAWAY (milliseconds). + * Gives the gateway (Envoy/Higress) time to complete drain setup + * while keeping the migration window short. + */ + private static final long GRACEFUL_RECONNECT_DELAY_MS = 200; + private final AbstractNettyConnectionClient connectionClient; public NettyConnectionHandler(AbstractNettyConnectionClient connectionClient) { @@ -55,18 +77,85 @@ public void onGoAway(Object channel) { if (Boolean.TRUE.equals(attr.get())) { return; } - attr.set(true); - if (connectionClient != null) { - connectionClient.setAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY, true); - connectionClient.onGoaway(nettyChannel); - } + if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Channel {} go away ,schedule reconnect", nettyChannel); + LOGGER.debug(String.format( + "Received GOAWAY on %s -> %s, initiating graceful migration (delay=%dms)", + nettyChannel.localAddress(), nettyChannel.remoteAddress(), GRACEFUL_RECONNECT_DELAY_MS)); + } + + if (connectionClient.isClosed()) { + LOGGER.info("The client has been closed and will not reconnect."); + return; + } + + // Use connectivityExecutor (a dedicated scheduler) instead of the Netty I/O + // EventLoop. doConnect() internally awaits the connect promise via + // awaitUninterruptibly(getConnectTimeout()), which would block the I/O thread + // and stall in-flight requests on this very channel. + final ScheduledExecutorService scheduler = connectionClient.getConnectivityExecutor(); + scheduler.schedule( + () -> attemptGracefulMigration(nettyChannel, scheduler, true), + GRACEFUL_RECONNECT_DELAY_MS, + TimeUnit.MILLISECONDS); + } + + /** + * Try to establish a new connection on behalf of {@link #onGoAway}. + * + *

On success, {@link AbstractNettyConnectionClient#onConnected} swaps the channel + * reference atomically (closeFuture CAS in initBootstrap protects against + * the old channel's close handler wiping the new reference). + * + *

On failure, the first attempt schedules one extra retry through the + * same connectivity executor. The second failure falls back to + * {@link AbstractNettyConnectionClient#onGoaway}, which nulls the channel and lets + * the connectivity scheduler take over recovery. + */ + private void attemptGracefulMigration( + final Channel nettyChannel, final ScheduledExecutorService scheduler, final boolean canRetry) { + if (connectionClient.isClosed()) { + return; + } + try { + connectionClient.doConnect(); + // Success: onConnected() will close old channel and set new one. + // The closeFuture CAS in initBootstrap() ensures no race condition. + } catch (Throwable e) { + if (canRetry) { + long retryDelay = Math.max(connectionClient.getReconnectDuration(), GRACEFUL_RECONNECT_DELAY_MS); + LOGGER.warn( + TRANSPORT_FAILED_RECONNECT, + "", + "", + String.format( + "Graceful migration attempt failed for %s, scheduling one retry in %dms", + nettyChannel.remoteAddress(), retryDelay), + e); + scheduler.schedule( + () -> attemptGracefulMigration(nettyChannel, scheduler, false), + retryDelay, + TimeUnit.MILLISECONDS); + return; + } + // Graceful migration failed twice (e.g., persistent mTLS timeout). + // Fallback: null the channel to let connectivity-scheduler take over. + LOGGER.error( + TRANSPORT_FAILED_RECONNECT, + "", + "", + String.format( + "Graceful migration failed for %s, falling back to scheduler recovery", + nettyChannel.remoteAddress()), + e); + connectionClient.onGoaway(nettyChannel); } - reconnect(nettyChannel); } + /** + * Reconnect with longer delay - used for unexpected disconnects (non-GOAWAY). + */ @Override public void reconnect(Object channel) { if (!(channel instanceof Channel)) { @@ -76,7 +165,7 @@ public void reconnect(Object channel) { LOGGER.debug("Connection:{} is reconnecting, attempt={}", connectionClient, 1); } if (connectionClient.isClosed()) { - LOGGER.info("The connection {} has been closed and will not reconnect", connectionClient); + LOGGER.info("The client has been closed and will not reconnect."); return; } connectionClient.scheduleReconnect(1, TimeUnit.SECONDS); @@ -106,13 +195,17 @@ public void channelActive(ChannelHandlerContext ctx) { public void channelInactive(ChannelHandlerContext ctx) throws Exception { super.channelInactive(ctx); Channel ch = ctx.channel(); - NettyChannel channel = NettyChannel.getOrAddChannel(ch, connectionClient.getUrl(), connectionClient); + // Look up the cached NettyChannel without allocating a new one. + // The channel is already inactive at this point, so getOrAddChannel would + // construct a transient unregistered NettyChannel only to log it. + NettyChannel channel = NettyChannel.getChannelIfPresent(ch); try { Attribute goawayAttr = ch.attr(GO_AWAY_KEY); if (!Boolean.TRUE.equals(goawayAttr.get())) { + // Only reconnect for unexpected disconnects (not GOAWAY). + // GOAWAY disconnects are already handled by the graceful migration above. reconnect(ch); } - if (LOGGER.isInfoEnabled() && channel != null) { LOGGER.info( "The connection {} of {} -> {} is disconnected.", diff --git a/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/TripleGoAwayTest.java b/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/TripleGoAwayTest.java new file mode 100644 index 000000000000..91ab33ab80b5 --- /dev/null +++ b/dubbo-remoting/dubbo-remoting-netty4/src/test/java/org/apache/dubbo/remoting/transport/netty4/TripleGoAwayTest.java @@ -0,0 +1,98 @@ +/* + * 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.dubbo.remoting.transport.netty4; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.constants.CommonConstants; +import org.apache.dubbo.common.utils.NetUtils; +import org.apache.dubbo.config.ApplicationConfig; +import org.apache.dubbo.config.context.ConfigManager; +import org.apache.dubbo.remoting.api.connection.AbstractConnectionClient; +import org.apache.dubbo.remoting.api.connection.ConnectionManager; +import org.apache.dubbo.remoting.api.connection.MultiplexProtocolConnectionManager; +import org.apache.dubbo.remoting.api.pu.DefaultPuHandler; +import org.apache.dubbo.rpc.model.ApplicationModel; +import org.apache.dubbo.rpc.model.ModuleModel; + +import java.time.Duration; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_MANAGEMENT_MODE_DEFAULT; +import static org.awaitility.Awaitility.await; + +public class TripleGoAwayTest { + + private static URL url; + + private static NettyPortUnificationServer server; + + private static ConnectionManager connectionManager; + + @BeforeAll + public static void init() throws Throwable { + int port = NetUtils.getAvailablePort(); + url = URL.valueOf("tri://127.0.0.1:" + port + "?foo=bar"); + ApplicationModel applicationModel = ApplicationModel.defaultModel(); + ApplicationConfig applicationConfig = new ApplicationConfig("provider-app"); + applicationConfig.setExecutorManagementMode(EXECUTOR_MANAGEMENT_MODE_DEFAULT); + applicationModel.getApplicationConfigManager().setApplication(applicationConfig); + ConfigManager configManager = new ConfigManager(applicationModel); + configManager.setApplication(applicationConfig); + configManager.getApplication(); + applicationModel.setConfigManager(configManager); + url = url.setScopeModel(applicationModel); + ModuleModel moduleModel = applicationModel.getDefaultModule(); + url = url.putAttribute(CommonConstants.SCOPE_MODEL, moduleModel); + server = new NettyPortUnificationServer(url, new DefaultPuHandler()); + server.bind(); + connectionManager = url.getOrDefaultFrameworkModel() + .getExtensionLoader(ConnectionManager.class) + .getExtension(MultiplexProtocolConnectionManager.NAME); + } + + @AfterAll + public static void close() { + try { + server.close(); + } catch (Throwable e) { + // ignored + } + } + + @Test + void testGoawayCausesNoChannelFailure() { + final AbstractConnectionClient connectionClient = connectionManager.connect(url, new DefaultPuHandler()); + Assertions.assertTrue( + connectionClient.isAvailable(), "Precondition: connection should be available before GOAWAY"); + io.netty.channel.Channel channelBeforeGoaway = connectionClient.getChannel(true); + NettyConnectionHandler handler = new NettyConnectionHandler((AbstractNettyConnectionClient) connectionClient); + handler.onGoAway(channelBeforeGoaway); + io.netty.channel.Channel channelAfterGoaway = connectionClient.getChannel(true); + Assertions.assertNotNull( + channelAfterGoaway, + "Channel should NOT be null immediately after GOAWAY " + + "— old channel should stay alive until new one is ready"); + await().atMost(Duration.ofSeconds(10)).until(() -> connectionClient.isAvailable()); + Assertions.assertTrue( + connectionClient.isAvailable(), "Connection should be available after GOAWAY migration completes"); + connectionClient.close(); + } +} diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleGoAwayHandler.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleGoAwayHandler.java index 6eb8fb7515ab..63671e062fb3 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleGoAwayHandler.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/transport/TripleGoAwayHandler.java @@ -26,6 +26,13 @@ import io.netty.handler.codec.http2.Http2GoAwayFrame; import io.netty.util.ReferenceCountUtil; +/** + * Handles HTTP/2 GOAWAY frames on the Triple protocol client side. + * + *

Logs the GOAWAY errorCode and lastStreamId for diagnostics, then delegates + * to {@link ConnectionHandler#onGoAway} which performs graceful connection migration + * (keeping old channel alive until new connection is established). + */ public class TripleGoAwayHandler extends ChannelDuplexHandler { private static final Logger logger = LoggerFactory.getLogger(TripleGoAwayHandler.class); @@ -35,12 +42,20 @@ public TripleGoAwayHandler() {} @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (msg instanceof Http2GoAwayFrame) { + Http2GoAwayFrame goAwayFrame = (Http2GoAwayFrame) msg; final ConnectionHandler connectionHandler = (ConnectionHandler) ctx.pipeline().get(Constants.CONNECTION_HANDLER_NAME); + if (logger.isInfoEnabled()) { - logger.info("Receive go away frame of " + ctx.channel().localAddress() + " -> " - + ctx.channel().remoteAddress() + " and will reconnect later."); + logger.info(String.format( + "Received GOAWAY frame: %s -> %s, errorCode=%d, lastStreamId=%d. " + + "Initiating graceful connection migration.", + ctx.channel().localAddress(), + ctx.channel().remoteAddress(), + goAwayFrame.errorCode(), + goAwayFrame.lastStreamId())); } + connectionHandler.onGoAway(ctx.channel()); ReferenceCountUtil.release(msg); return;