You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "BackendArchitectX (via GitHub)" <gi...@apache.org> on 2026/09/09 08:47:21 UTC
[PR] [rpc] Handle endpoint changes for cached server connections [fluss]
BackendArchitectX opened a new pull request, #4263:
URL: https://github.com/apache/fluss/pull/4263
<!--
Generated-by: ChatGPT (GPT-5.6 Sol) following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
Linked issue: close #4256
`NettyClient` currently caches connections only by server UID. During a rolling upgrade, the same TabletServer UID may be advertised with a different host or port while the previous endpoint remains reachable.
In that case, even after metadata is refreshed with the new endpoint, connection lookup can continue returning the connection to the previous endpoint. Requests may therefore keep reaching the wrong TabletServer.
This change makes the physical RPC connection identity endpoint-aware by using server UID, host, and port.
This is related to #4216, which also handles same-UID endpoint changes but replaces the existing connection. This change keeps connections to different physical endpoints independently keyed so the RPC layer does not have to decide which reachable endpoint is newer.
### Brief change log
- Key cached Netty connections by server UID, host, and port instead of server UID alone.
- Allow different physical endpoints for the same logical server UID to use independent connections.
- Keep `disconnect(serverUid)` as a logical-server operation by closing all connections associated with that UID.
- Preserve UID-level readiness semantics.
- Use endpoint-specific connection metric identities so multiple connections sharing a UID do not overwrite each other's metric bookkeeping.
- Add regression coverage verifying that a changed endpoint for the same UID is actually used while the previous endpoint remains reachable.
- Add lifecycle coverage verifying that disconnecting a UID closes all associated endpoint connections.
### Tests
- `.\mvnw.cmd test -Dtest=NettyClientTest -pl fluss-rpc`
- 9 tests passed
- 0 failures
- 0 errors
- `.\mvnw.cmd verify -pl fluss-rpc -Dtest="!ApiErrorTest"`
- 89 tests passed
- 0 failures
- 0 errors
- 2 skipped
- Checkstyle passed with 0 violations.
- `git diff --check` passed.
A full `fluss-rpc` verify on Windows also encounters the existing `ApiErrorTest#testStringifyException` line-ending assertion, which expects LF while the Windows JVM stack trace uses CRLF. This test is unrelated to this change.
The new regression test intentionally isolates the transport-level condition: two reachable TabletServers use the same UID with different endpoints, and the test verifies that a request using the updated `ServerNode` reaches the updated endpoint rather than the previously cached one.
### API and Format
No public API, RPC protocol, or storage format changes.
### Documentation
No documentation changes. This fixes existing RPC connection behavior when a server endpoint changes.
### Generative AI disclosure
Yes — ChatGPT (GPT-5.6 Sol) was used for root-cause analysis, implementation review, and test planning.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on code in PR #4263:
URL: https://github.com/apache/fluss/pull/4263#discussion_r3980357600
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -146,11 +157,14 @@ public CompletableFuture<Void> disconnect(String serverUid) {
@Override
public boolean isReady(String serverUid) {
checkArgument(!isClosed, "Netty client is closed.");
- ServerConnection connection = connections.get(serverUid);
- if (connection == null) {
- return false;
+
+ for (Map.Entry<ConnectionKey, ServerConnection> entry : connections.entrySet()) {
Review Comment:
Thanks @gyang94. Updated. `isReady` now takes `ServerNode` and checks the exact `serverUid + host + port` connection, so readiness of another endpoint sharing the same UID no longer affects the result. I also added coverage verifying that one endpoint can be ready while another endpoint with the same UID is not yet connected.
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -130,11 +131,21 @@ public boolean connect(ServerNode node) {
public CompletableFuture<Void> disconnect(String serverUid) {
LOG.debug("Disconnecting from server {}.", serverUid);
checkArgument(!isClosed, "Netty client is closed.");
- ServerConnection connection = connections.remove(serverUid);
- if (connection != null) {
- return connection.close();
+
+ List<CompletableFuture<Void>> shutdownFutures = new ArrayList<>();
+
+ for (Map.Entry<ConnectionKey, ServerConnection> entry : connections.entrySet()) {
+ if (entry.getKey().belongsTo(serverUid)
+ && connections.remove(entry.getKey(), entry.getValue())) {
Review Comment:
Thanks @gyang94. Added `disconnect(ServerNode node)` for disconnecting a specific endpoint. I kept `disconnect(String serverUid)` as the logical-server operation that closes all endpoint connections associated with the UID. Added coverage for both endpoint-specific and UID-wide disconnect behavior.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-6340339437
Hi @loserwang1024,
I've rebased and consolidated the PR into a single commit (`29cbc8789`) as requested. The focused regression tests pass locally (25 tests, 0 failures).
The GitHub Actions workflows are currently awaiting maintainer approval. Could you or another maintainer please approve them when convenient so CI can run?
Thank you for your review and support.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-6489026895
I'll merge this first, then update the Mockito-based tests.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #4263:
URL: https://github.com/apache/fluss/pull/4263#discussion_r4398824639
##########
fluss-client/src/test/java/org/apache/fluss/client/metadata/MetadataUpdaterTest.java:
##########
@@ -39,12 +40,19 @@
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.buildMetadataResponse;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
Review Comment:
One small suggestion: please avoid Mockito, as it makes tests harder to maintain over time. Let’s use dedicated test implementations instead.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on code in PR #4263:
URL: https://github.com/apache/fluss/pull/4263#discussion_r4347837365
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -66,10 +67,10 @@ public final class NettyClient implements RpcClient {
private final EventLoopGroup eventGroup;
/**
- * Managed connections to Netty servers. The key is the server uid (e.g., "cs-2", "ts-3"), the
- * value is the connection.
+ * Managed connections to Netty servers. Connections are identified by server uid, host and
+ * port.
*/
- private final Map<String, ServerConnection> connections;
+ private final Map<ConnectionKey, ServerConnection> connections;
Review Comment:
Thanks @loserwang1024 for the detailed review.
Addressed in commit `6e66c9f9f`:
- Added `ConnectionKey` Javadoc explaining why endpoints sharing a UID must coexist and how idle connections are cleaned up.
- Updated the PR description to document the 10-minute default idle timeout and temporary connection overhead during rolling upgrades.
- Updated `MetadataUpdater` to disconnect only the failed bootstrap endpoint, with regression tests covering retry and failover.
Validation: 23 focused tests passed with no failures.
Please let me know if any further changes are needed.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-5600025804
@wuchong @loserwang1024 @swuferhong @luoyuxia, would appreciate your thoughts on the connection-handling approach in this PR. It overlaps with the same-UID endpoint handling discussed in #4216, but keeps physical connections keyed by UID + host + port instead of replacing the existing UID connection. Any feedback on the preferred direction would be very helpful.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "gyang94 (via GitHub)" <gi...@apache.org>.
gyang94 commented on code in PR #4263:
URL: https://github.com/apache/fluss/pull/4263#discussion_r3979313861
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -130,11 +131,21 @@ public boolean connect(ServerNode node) {
public CompletableFuture<Void> disconnect(String serverUid) {
LOG.debug("Disconnecting from server {}.", serverUid);
checkArgument(!isClosed, "Netty client is closed.");
- ServerConnection connection = connections.remove(serverUid);
- if (connection != null) {
- return connection.close();
+
+ List<CompletableFuture<Void>> shutdownFutures = new ArrayList<>();
+
+ for (Map.Entry<ConnectionKey, ServerConnection> entry : connections.entrySet()) {
+ if (entry.getKey().belongsTo(serverUid)
+ && connections.remove(entry.getKey(), entry.getValue())) {
Review Comment:
Here we remove all connections for a serverUid, which is fine.
Do we need a function `disconnect(ServerNode node)` to disconnect a specific connection to a `serverId + host` ?
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -146,11 +157,14 @@ public CompletableFuture<Void> disconnect(String serverUid) {
@Override
public boolean isReady(String serverUid) {
checkArgument(!isClosed, "Netty client is closed.");
- ServerConnection connection = connections.get(serverUid);
- if (connection == null) {
- return false;
+
+ for (Map.Entry<ConnectionKey, ServerConnection> entry : connections.entrySet()) {
Review Comment:
Feel a little bit strange here. One serverUid may have multiple ServerConnections, a connection is ready, this method will return true.
Will it be better to change this function to `isReady(ServerNode node)` ? to check a connection to a `server + host` is ready or not.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-6276590280
Hi @gyang94,
I've resolved the merge conflicts and updated the PR description with the latest validation results.
- 13 NettyClient tests passed.
- 159 RPC verification tests completed with 0 failures and 2 skipped.
- 24 cross-module RPC, Server, and Client tests passed with 0 failures.
- Checkstyle and Spotless passed.
The PR is now conflict-free. All four GitHub Actions workflows are still awaiting maintainer approval.
Could you please help approve the workflows and review the updated changes when convenient?
Thanks!
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-6291911617
Hi @gyang94,
Thank you for the review and approval.
I've resolved the merge conflicts and updated the PR description with the latest validation results:
- NettyClient: 12/12 tests passed.
- RPC verification: 156 tests, 0 failures, 2 skipped.
- Cross-module regression: 23/23 tests passed.
- All 11 Maven reactor modules completed successfully.
- Checkstyle and Spotless passed.
Could a maintainer please approve the four pending GitHub Actions workflows when convenient?
Thanks!
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "gyang94 (via GitHub)" <gi...@apache.org>.
gyang94 commented on code in PR #4263:
URL: https://github.com/apache/fluss/pull/4263#discussion_r3979294275
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -146,11 +157,14 @@ public CompletableFuture<Void> disconnect(String serverUid) {
@Override
public boolean isReady(String serverUid) {
checkArgument(!isClosed, "Netty client is closed.");
- ServerConnection connection = connections.get(serverUid);
- if (connection == null) {
- return false;
+
+ for (Map.Entry<ConnectionKey, ServerConnection> entry : connections.entrySet()) {
Review Comment:
Feel a little bit strange here. One serverUid may have multiple ServerConnections, any one of connections is ready, this method will return true.
Will it be better to change this function to `isReady(ServerNode node)` ? to check a connection to a `server + host` is ready or not.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong merged PR #4263:
URL: https://github.com/apache/fluss/pull/4263
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "BackendArchitectX (via GitHub)" <gi...@apache.org>.
BackendArchitectX commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-6087894620
Hi @gyang94, I've completed the latest updates to this PR.
Local validation is successful across RPC, Server, and Client: all 23 selected tests passed, with Checkstyle and Spotless checks passing.
The GitHub CI and license workflows are currently marked `action_required`. Could you please help approve the workflow runs and review the updated changes when convenient?
Thanks!
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "gyang94 (via GitHub)" <gi...@apache.org>.
gyang94 commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-5619001925
@BackendArchitectX Thanks! Using `serverId + host + port` as the composite key for the cache seems like a good approach. It will help differentiate servers with both legacy and new host addresses.
I am thinking that do we need to change the method in `NettyClient` like `isReady(String serverId)` and `disconnect(String serverId)` may change to `isReady(ServerNode node)` and `disconnect(ServerNode node)`, to match the new key semantics. What do you think?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "gyang94 (via GitHub)" <gi...@apache.org>.
gyang94 commented on PR #4263:
URL: https://github.com/apache/fluss/pull/4263#issuecomment-6287178436
@BackendArchitectX Thanks. LGTM.
btw, there are conflicts need to resolve.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [rpc] Handle endpoint changes for cached server connections [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4263:
URL: https://github.com/apache/fluss/pull/4263#discussion_r4332284753
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -66,10 +67,10 @@ public final class NettyClient implements RpcClient {
private final EventLoopGroup eventGroup;
/**
- * Managed connections to Netty servers. The key is the server uid (e.g., "cs-2", "ts-3"), the
- * value is the connection.
+ * Managed connections to Netty servers. Connections are identified by server uid, host and
+ * port.
*/
- private final Map<String, ServerConnection> connections;
+ private final Map<ConnectionKey, ServerConnection> connections;
Review Comment:
Please document why two endpoints for one uid coexist. Nothing in ConnectionKey (:254) explains it, so the next reader will try to collapse it into evict-and-replace. Two points worth a javadoc
ServerNode has no epoch/version, so the transport layer can't tell which endpoint is newer; "incoming wins" breaks when callers hold different metadata vintages concurrently (MetadataUpdater.cluster is swapped wholesale, readers are unsynchronized, and KvBatchScanner.gateway stays pinned for a whole scan). Those two would evict each other per request.
Stale entries are reclaimed by netty.connection.max-idle-time (10 min default) via IdleStateHandler → NettyClientHandler:163. Bounded, but it does mean ~2x connections during a rolling upgrade — worth stating in the PR description.
##########
fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/client/NettyClient.java:
##########
@@ -130,11 +131,21 @@ public boolean connect(ServerNode node) {
public CompletableFuture<Void> disconnect(String serverUid) {
LOG.debug("Disconnecting from server {}.", serverUid);
checkArgument(!isClosed, "Netty client is closed.");
- ServerConnection connection = connections.remove(serverUid);
- if (connection != null) {
- return connection.close();
+
+ List<CompletableFuture<Void>> shutdownFutures = new ArrayList<>();
+
+ for (Map.Entry<ConnectionKey, ServerConnection> entry : connections.entrySet()) {
+ if (entry.getKey().belongsTo(serverUid)
+ && connections.remove(entry.getKey(), entry.getValue())) {
Review Comment:
disconnect(ServerNode) currently has no production caller. MetadataUpdater:388/:434 are candidates — bootstrap nodes all share uid "ts--1" (ServerNode:56-60 maps UNKNOWN to the ts- branch), so disconnect(uid) there is broader than the intent.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org