fresh-borzoni commented on code in PR #3390: URL: https://github.com/apache/fluss/pull/3390#discussion_r3321608857
########## fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java: ########## @@ -0,0 +1,159 @@ +/* + * 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.fluss.rpc; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.RetriableException; +import org.apache.fluss.utils.ExceptionUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.concurrent.CompletableFuture; + +/** + * A proxy that wraps an existing {@link RpcGateway} proxy and adds automatic retry with metadata + * refresh on retriable (network) errors. + * + * <p>This is designed to solve the stale metadata problem where cached server addresses become + * invalid (e.g., during rolling upgrades in Kubernetes). When an RPC call fails with a {@link + * RetriableException}, this proxy triggers a metadata refresh callback and retries the request with + * potentially updated server addresses. + * + * <p>The retry flow for a cluster with N stale tablet servers: + * + * <ol> + * <li>RPC fails with {@link RetriableException} (e.g., connection refused to stale IP) + * <li>Metadata refresh is triggered, which marks the failed server as unavailable + * <li>After N failed refreshes, all servers are marked unavailable, triggering re-initialization + * from bootstrap servers + * <li>The next retry succeeds with the refreshed server addresses + * </ol> + */ +@Internal +public class RetryableGatewayClientProxy implements InvocationHandler { + + private static final Logger LOG = LoggerFactory.getLogger(RetryableGatewayClientProxy.class); + + private final Object delegate; + private final Runnable metadataRefreshAction; + private final int maxRetries; + + RetryableGatewayClientProxy(Object delegate, Runnable metadataRefreshAction, int maxRetries) { + this.delegate = delegate; + this.metadataRefreshAction = metadataRefreshAction; + this.maxRetries = maxRetries; + } + + /** + * Creates a retryable proxy wrapping an existing gateway proxy. On {@link RetriableException}, + * the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC call. + * + * @param delegate the underlying gateway proxy to wrap + * @param metadataRefreshAction callback to refresh metadata (e.g., update cluster info) + * @param maxRetries maximum number of retries before propagating the error + * @param gatewayClass the gateway interface class + * @param <T> the gateway type + * @return a retryable gateway proxy + */ + public static <T extends RpcGateway> T createRetryableGatewayProxy( + T delegate, Runnable metadataRefreshAction, int maxRetries, Class<T> gatewayClass) { + ClassLoader classLoader = gatewayClass.getClassLoader(); + + @SuppressWarnings("unchecked") + T proxy = + (T) + Proxy.newProxyInstance( + classLoader, + new Class<?>[] {gatewayClass}, + new RetryableGatewayClientProxy( + delegate, metadataRefreshAction, maxRetries)); + return proxy; + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + return invokeWithRetry(method, args, 0); + } + + @SuppressWarnings("unchecked") + private <T> CompletableFuture<T> invokeWithRetry(Method method, Object[] args, int attempt) { + CompletableFuture<T> future; + try { + future = (CompletableFuture<T>) method.invoke(delegate, args); + } catch (InvocationTargetException e) { + CompletableFuture<T> failed = new CompletableFuture<>(); + failed.completeExceptionally(e.getCause()); + return failed; + } catch (Exception e) { + CompletableFuture<T> failed = new CompletableFuture<>(); + failed.completeExceptionally(e); + return failed; + } + + CompletableFuture<T> resultFuture = new CompletableFuture<>(); + future.whenComplete( + (result, throwable) -> { + if (throwable == null) { + resultFuture.complete(result); + return; + } + Throwable cause = ExceptionUtils.stripCompletionException(throwable); + if (!(cause instanceof RetriableException) || attempt >= maxRetries) { + resultFuture.completeExceptionally(cause); + return; + } + LOG.warn( + "RPC call {} failed with retriable error (attempt {}/{}), " + + "refreshing metadata and retrying.", + method.getName(), + attempt + 1, + maxRetries, + cause); + // Run metadata refresh and retry on a separate thread to avoid + // blocking Netty IO threads that may complete the failed future. + CompletableFuture.runAsync( Review Comment: do we want some backoff? I mean 3 retries fire in milliseconds, seems wasteful on slow DNS or restarting pods. ########## fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java: ########## @@ -139,13 +140,21 @@ public class FlussAdmin implements Admin { private final AdminReadOnlyGateway readOnlyGateway; private final MetadataUpdater metadataUpdater; + private static final int READ_ONLY_GATEWAY_MAX_RETRIES = 3; Review Comment: With maxRetries=3, bootstrap reinit needs 4 refreshes. You only get 3 per request. Shall we loop inside updateMetadata until either success or null-triggered bootstrap? ########## fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java: ########## @@ -0,0 +1,159 @@ +/* + * 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.fluss.rpc; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.RetriableException; +import org.apache.fluss.utils.ExceptionUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.concurrent.CompletableFuture; + +/** + * A proxy that wraps an existing {@link RpcGateway} proxy and adds automatic retry with metadata + * refresh on retriable (network) errors. + * + * <p>This is designed to solve the stale metadata problem where cached server addresses become + * invalid (e.g., during rolling upgrades in Kubernetes). When an RPC call fails with a {@link + * RetriableException}, this proxy triggers a metadata refresh callback and retries the request with + * potentially updated server addresses. + * + * <p>The retry flow for a cluster with N stale tablet servers: + * + * <ol> + * <li>RPC fails with {@link RetriableException} (e.g., connection refused to stale IP) + * <li>Metadata refresh is triggered, which marks the failed server as unavailable + * <li>After N failed refreshes, all servers are marked unavailable, triggering re-initialization + * from bootstrap servers + * <li>The next retry succeeds with the refreshed server addresses Review Comment: ditto: only true when maxRetries > cluster_size ########## fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java: ########## @@ -0,0 +1,159 @@ +/* + * 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.fluss.rpc; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.RetriableException; +import org.apache.fluss.utils.ExceptionUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.concurrent.CompletableFuture; + +/** + * A proxy that wraps an existing {@link RpcGateway} proxy and adds automatic retry with metadata + * refresh on retriable (network) errors. + * + * <p>This is designed to solve the stale metadata problem where cached server addresses become + * invalid (e.g., during rolling upgrades in Kubernetes). When an RPC call fails with a {@link + * RetriableException}, this proxy triggers a metadata refresh callback and retries the request with + * potentially updated server addresses. + * + * <p>The retry flow for a cluster with N stale tablet servers: + * + * <ol> + * <li>RPC fails with {@link RetriableException} (e.g., connection refused to stale IP) + * <li>Metadata refresh is triggered, which marks the failed server as unavailable + * <li>After N failed refreshes, all servers are marked unavailable, triggering re-initialization + * from bootstrap servers + * <li>The next retry succeeds with the refreshed server addresses + * </ol> + */ +@Internal +public class RetryableGatewayClientProxy implements InvocationHandler { + + private static final Logger LOG = LoggerFactory.getLogger(RetryableGatewayClientProxy.class); + + private final Object delegate; + private final Runnable metadataRefreshAction; + private final int maxRetries; + + RetryableGatewayClientProxy(Object delegate, Runnable metadataRefreshAction, int maxRetries) { + this.delegate = delegate; + this.metadataRefreshAction = metadataRefreshAction; + this.maxRetries = maxRetries; + } + + /** + * Creates a retryable proxy wrapping an existing gateway proxy. On {@link RetriableException}, + * the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC call. + * + * @param delegate the underlying gateway proxy to wrap + * @param metadataRefreshAction callback to refresh metadata (e.g., update cluster info) + * @param maxRetries maximum number of retries before propagating the error + * @param gatewayClass the gateway interface class + * @param <T> the gateway type + * @return a retryable gateway proxy + */ + public static <T extends RpcGateway> T createRetryableGatewayProxy( + T delegate, Runnable metadataRefreshAction, int maxRetries, Class<T> gatewayClass) { + ClassLoader classLoader = gatewayClass.getClassLoader(); + + @SuppressWarnings("unchecked") + T proxy = + (T) + Proxy.newProxyInstance( + classLoader, + new Class<?>[] {gatewayClass}, + new RetryableGatewayClientProxy( + delegate, metadataRefreshAction, maxRetries)); + return proxy; + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + return invokeWithRetry(method, args, 0); + } + + @SuppressWarnings("unchecked") + private <T> CompletableFuture<T> invokeWithRetry(Method method, Object[] args, int attempt) { + CompletableFuture<T> future; + try { + future = (CompletableFuture<T>) method.invoke(delegate, args); + } catch (InvocationTargetException e) { + CompletableFuture<T> failed = new CompletableFuture<>(); + failed.completeExceptionally(e.getCause()); + return failed; + } catch (Exception e) { + CompletableFuture<T> failed = new CompletableFuture<>(); + failed.completeExceptionally(e); + return failed; + } + + CompletableFuture<T> resultFuture = new CompletableFuture<>(); + future.whenComplete( + (result, throwable) -> { + if (throwable == null) { + resultFuture.complete(result); + return; + } + Throwable cause = ExceptionUtils.stripCompletionException(throwable); + if (!(cause instanceof RetriableException) || attempt >= maxRetries) { + resultFuture.completeExceptionally(cause); + return; + } + LOG.warn( + "RPC call {} failed with retriable error (attempt {}/{}), " + + "refreshing metadata and retrying.", + method.getName(), + attempt + 1, + maxRetries, + cause); + // Run metadata refresh and retry on a separate thread to avoid + // blocking Netty IO threads that may complete the failed future. + CompletableFuture.runAsync( Review Comment: Every retry here fires its own updateMetadata call, and that method's synchronized(this) block is the same paths use to refresh leader info. Example: during a rolling upgrade, N concurrent failing admin calls × 3 retries all queue up behind one lock, and the data plane's refreshes wait in the same line. Could we share one in-flight refresh across concurrent retriers? ########## fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java: ########## @@ -139,13 +140,21 @@ public class FlussAdmin implements Admin { private final AdminReadOnlyGateway readOnlyGateway; private final MetadataUpdater metadataUpdater; + private static final int READ_ONLY_GATEWAY_MAX_RETRIES = 3; + public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { this.gateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getCoordinatorServer, client, AdminGateway.class); - this.readOnlyGateway = + AdminGateway rawReadOnlyGateway = Review Comment: Shall we add TODO for writes, since they are still broken? ########## fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java: ########## @@ -139,13 +140,21 @@ public class FlussAdmin implements Admin { private final AdminReadOnlyGateway readOnlyGateway; private final MetadataUpdater metadataUpdater; + private static final int READ_ONLY_GATEWAY_MAX_RETRIES = 3; Review Comment: Shall we make it a ConfigOption to make more operator-friendly? ########## fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java: ########## @@ -139,13 +140,21 @@ public class FlussAdmin implements Admin { private final AdminReadOnlyGateway readOnlyGateway; private final MetadataUpdater metadataUpdater; + private static final int READ_ONLY_GATEWAY_MAX_RETRIES = 3; + public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { this.gateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getCoordinatorServer, client, AdminGateway.class); - this.readOnlyGateway = + AdminGateway rawReadOnlyGateway = GatewayClientProxy.createGatewayProxy( metadataUpdater::getRandomTabletServer, client, AdminGateway.class); Review Comment: AdminReadOnlyGateway.class? ########## fluss-rpc/src/main/java/org/apache/fluss/rpc/RetryableGatewayClientProxy.java: ########## @@ -0,0 +1,159 @@ +/* + * 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.fluss.rpc; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.RetriableException; +import org.apache.fluss.utils.ExceptionUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.concurrent.CompletableFuture; + +/** + * A proxy that wraps an existing {@link RpcGateway} proxy and adds automatic retry with metadata + * refresh on retriable (network) errors. + * + * <p>This is designed to solve the stale metadata problem where cached server addresses become + * invalid (e.g., during rolling upgrades in Kubernetes). When an RPC call fails with a {@link + * RetriableException}, this proxy triggers a metadata refresh callback and retries the request with + * potentially updated server addresses. + * + * <p>The retry flow for a cluster with N stale tablet servers: + * + * <ol> + * <li>RPC fails with {@link RetriableException} (e.g., connection refused to stale IP) + * <li>Metadata refresh is triggered, which marks the failed server as unavailable + * <li>After N failed refreshes, all servers are marked unavailable, triggering re-initialization + * from bootstrap servers + * <li>The next retry succeeds with the refreshed server addresses + * </ol> + */ +@Internal +public class RetryableGatewayClientProxy implements InvocationHandler { + + private static final Logger LOG = LoggerFactory.getLogger(RetryableGatewayClientProxy.class); + + private final Object delegate; + private final Runnable metadataRefreshAction; + private final int maxRetries; + + RetryableGatewayClientProxy(Object delegate, Runnable metadataRefreshAction, int maxRetries) { + this.delegate = delegate; + this.metadataRefreshAction = metadataRefreshAction; + this.maxRetries = maxRetries; + } + + /** + * Creates a retryable proxy wrapping an existing gateway proxy. On {@link RetriableException}, + * the proxy will invoke {@code metadataRefreshAction} and retry the failed RPC call. + * + * @param delegate the underlying gateway proxy to wrap + * @param metadataRefreshAction callback to refresh metadata (e.g., update cluster info) + * @param maxRetries maximum number of retries before propagating the error + * @param gatewayClass the gateway interface class + * @param <T> the gateway type + * @return a retryable gateway proxy + */ + public static <T extends RpcGateway> T createRetryableGatewayProxy( + T delegate, Runnable metadataRefreshAction, int maxRetries, Class<T> gatewayClass) { + ClassLoader classLoader = gatewayClass.getClassLoader(); + + @SuppressWarnings("unchecked") + T proxy = + (T) + Proxy.newProxyInstance( + classLoader, + new Class<?>[] {gatewayClass}, + new RetryableGatewayClientProxy( + delegate, metadataRefreshAction, maxRetries)); + return proxy; + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + return invokeWithRetry(method, args, 0); + } + + @SuppressWarnings("unchecked") + private <T> CompletableFuture<T> invokeWithRetry(Method method, Object[] args, int attempt) { + CompletableFuture<T> future; + try { + future = (CompletableFuture<T>) method.invoke(delegate, args); + } catch (InvocationTargetException e) { + CompletableFuture<T> failed = new CompletableFuture<>(); + failed.completeExceptionally(e.getCause()); + return failed; + } catch (Exception e) { + CompletableFuture<T> failed = new CompletableFuture<>(); + failed.completeExceptionally(e); + return failed; + } + + CompletableFuture<T> resultFuture = new CompletableFuture<>(); + future.whenComplete( + (result, throwable) -> { + if (throwable == null) { + resultFuture.complete(result); + return; + } + Throwable cause = ExceptionUtils.stripCompletionException(throwable); + if (!(cause instanceof RetriableException) || attempt >= maxRetries) { + resultFuture.completeExceptionally(cause); + return; + } + LOG.warn( + "RPC call {} failed with retriable error (attempt {}/{}), " + + "refreshing metadata and retrying.", + method.getName(), + attempt + 1, + maxRetries, + cause); + // Run metadata refresh and retry on a separate thread to avoid + // blocking Netty IO threads that may complete the failed future. + CompletableFuture.runAsync( Review Comment: runAsync without an executor uses ForkJoinPool.commonPool(), should we use a dedicated executor 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
