Skip to content

HDFS-17909. [ARR] AsyncRouterHandlerExecutors should use bounded queue #8448

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ZanderXu merged 5 commits into from
Jun 1, 2026
Merged

HDFS-17909. [ARR] AsyncRouterHandlerExecutors should use bounded queue #8448

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
7173986
HDFS-17909. [ARR] AsyncRouterHandlerExecutors should use bounded queue
kokonguyen191 Apr 13, 2026
546e117
HDFS-17909. Change UTs to use waitFor()
kokonguyen191 May 15, 2026
a35b37b
HDFS-17909. Minor code update
kokonguyen191 May 18, 2026
7e45c36
HDFS-17909. Compact catch block in RouterAsyncRpcClient#invokeMethod
kokonguyen191 May 19, 2026
4bfe147
HDFS-17909. Remove unused import
kokonguyen191 May 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions ...hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RBFConfigKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,9 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
public static final String DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY =
FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "handler.count";
public static final int DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT = 10;
public static final String DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE =
FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "queue.size";
public static final int DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT = 1000;
public static final String DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY =
FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "responder.count";
public static final int DFS_ROUTER_ASYNCRPC_RESPONDER_COUNT_DEFAULT = 10;
Expand Down
124 changes: 80 additions & 44 deletions ...fs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FEDERATION_RENAME_OPTION;
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FEDERATION_RENAME_OPTION_DEFAULT;
Expand Down Expand Up @@ -58,7 +60,7 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
Expand All @@ -70,7 +72,9 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -293,11 +297,11 @@ public class RouterRpcServer extends AbstractService implements ClientProtocol,
private RouterRenameOption routerRenameOption;
/** Schedule the router federation rename jobs. */
private BalanceProcedureScheduler fedRenameScheduler;
private boolean enableAsync;
private Map<String, Integer> nsAsyncHandlerCount = new ConcurrentHashMap<>();
private Map<String, ExecutorService> asyncRouterHandlerExecutors = new ConcurrentHashMap<>();
private final boolean enableAsync;
private final Map<String, ThreadPoolExecutor> asyncRouterHandlerExecutors =
new ConcurrentHashMap<>();
private ThreadPoolExecutor routerDefaultAsyncHandlerExecutor;
private ExecutorService routerAsyncResponderExecutor;
private ExecutorService routerDefaultAsyncHandlerExecutor;

/**
* Construct a router RPC server.
Expand Down Expand Up @@ -504,34 +508,60 @@ public RouterRpcServer(Configuration conf, Router router,
* @param configuration the configuration.
*/
public void initAsyncThreadPools(Configuration configuration) {
LOG.info("Begin initialize asynchronous handler and responder thread pool.");
initNsAsyncHandlerCount();
Set<String> allConfiguredNS = FederationUtil.getAllConfiguredNS(configuration);
Set<String> unassignedNS = new HashSet<>();
allConfiguredNS.add(CONCURRENT_NS);

Map<String, Integer> nsAsyncActiveHandlerCount = parseNsAsyncHandlerCount(configuration);
initAsyncHandlerThreadPools(configuration, allConfiguredNS, nsAsyncActiveHandlerCount);
initAsyncResponderThreadPools(configuration);
}

private void initAsyncHandlerThreadPools(Configuration configuration,
Set<String> allConfiguredNS, Map<String, Integer> nsAsyncHandlerCount) {
LOG.info("Initializing asynchronous handler thread pools");
int asyncQueueSize = configuration.getInt(DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE,
DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT);
if (asyncQueueSize < 1) {
throw new IllegalArgumentException("Async queue size must be at least 1");
}
int asyncHandlerCountDefault = configuration.getInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT);
Comment thread
kokonguyen191 marked this conversation as resolved.
if (asyncHandlerCountDefault < 1) {
throw new IllegalArgumentException("Async handler count must be at least 1");
}
for (String nsId : allConfiguredNS) {
int dedicatedHandlers = nsAsyncHandlerCount.getOrDefault(nsId, 0);
LOG.info("Dedicated handlers {} for ns {} ", dedicatedHandlers, nsId);
if (dedicatedHandlers > 0) {
initAsyncHandlerThreadPools4Ns(nsId, dedicatedHandlers);
LOG.info("Assigned {} async handlers to nsId {} ", dedicatedHandlers, nsId);
if (dedicatedHandlers <= 0) {
dedicatedHandlers = asyncHandlerCountDefault;
LOG.info("Use default async handler count {} for ns {} to init Executors.",
asyncHandlerCountDefault, nsId);
} else {
unassignedNS.add(nsId);
LOG.info("Dedicated handlers {} for ns {} to init Executors", dedicatedHandlers, nsId);
}
}

int asyncHandlerCountDefault = configuration.getInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT);
int finalDedicatedHandlers = dedicatedHandlers;
asyncRouterHandlerExecutors.computeIfAbsent(nsId,
id -> {
LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(asyncQueueSize);
return new ThreadPoolExecutor(finalDedicatedHandlers, finalDedicatedHandlers,
0L, TimeUnit.MILLISECONDS, queue,
new AsyncThreadFactory("Router Async Handler for " + nsId + " #"));
});
LOG.info("Assigned {} async handlers with queue size {} to nsId {}", dedicatedHandlers,
asyncQueueSize, nsId);
}

if (!unassignedNS.isEmpty()) {
LOG.warn("Async handler unassigned ns: {}", unassignedNS);
LOG.info("Use default async handler count {} for unassigned ns.", asyncHandlerCountDefault);
for (String nsId : unassignedNS) {
initAsyncHandlerThreadPools4Ns(nsId, asyncHandlerCountDefault);
}
if (routerDefaultAsyncHandlerExecutor == null) {
LOG.info("Initializing router default async handler executor, count={}",
asyncHandlerCountDefault);
LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(asyncQueueSize);
routerDefaultAsyncHandlerExecutor = new ThreadPoolExecutor(asyncHandlerCountDefault,
asyncHandlerCountDefault, 0L, TimeUnit.MILLISECONDS, queue,
new AsyncThreadFactory("Router Default Async Handler #"));
}
}

private void initAsyncResponderThreadPools(Configuration configuration) {
LOG.info("Initializing asynchronous responder thread pool.");
int asyncResponderCount = configuration.getInt(DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY,
DFS_ROUTER_ASYNCRPC_RESPONDER_COUNT_DEFAULT);
if (routerAsyncResponderExecutor == null) {
Expand All @@ -540,21 +570,17 @@ public void initAsyncThreadPools(Configuration configuration) {
asyncResponderCount, new AsyncThreadFactory("Router Async Responder #"));
}
AsyncRpcProtocolPBUtil.setAsyncResponderExecutor(routerAsyncResponderExecutor);

if (routerDefaultAsyncHandlerExecutor == null) {
LOG.info("init router async default executor handler count: {}", asyncHandlerCountDefault);
routerDefaultAsyncHandlerExecutor = Executors.newFixedThreadPool(
asyncHandlerCountDefault, new AsyncThreadFactory("Router Async Default Handler #"));
}
}

private void initNsAsyncHandlerCount() {
String configNsHandler = conf.get(DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY,
private Map<String, Integer> parseNsAsyncHandlerCount(Configuration config) {
String configNsHandler = config.get(DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY,
DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT);
Map<String, Integer> nsAsyncHandlerCount = new HashMap<>();
if (StringUtils.isEmpty(configNsHandler)) {
LOG.error(
"The value of config key: {} is empty. Will use default conf.",
LOG.info("No per-namespace async handler counts configured ({}). "
+ "Will use default handler count for all namespaces.",
DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY);
return nsAsyncHandlerCount;
}
String[] nsHandlers = configNsHandler.split(",");
for (String nsHandlerInfo : nsHandlers) {
Expand All @@ -567,11 +593,19 @@ private void initNsAsyncHandlerCount() {
}
nsAsyncHandlerCount.put(nsHandlerItems[0], Integer.parseInt(nsHandlerItems[1]));
}
return nsAsyncHandlerCount;
}

private void initAsyncHandlerThreadPools4Ns(String nsId, int dedicatedHandlers) {
asyncRouterHandlerExecutors.computeIfAbsent(nsId, id -> Executors.newFixedThreadPool(
dedicatedHandlers, new AsyncThreadFactory("Router Async Handler for " + id + " #")));
/**
* Returns the asynchronous handler executor for the specified namespace.
* If no executor is configured for the given namespace ID, returns the default executor.
* Requires async RPC pools to be initialized (async RPC enabled).
*
* @param nsId the namespace identifier
* @return the corresponding thread pool
*/
public ThreadPoolExecutor getAsyncExecutorForNamespace(String nsId) {
return asyncRouterHandlerExecutors.getOrDefault(nsId, routerDefaultAsyncHandlerExecutor);
}

/**
Expand Down Expand Up @@ -683,6 +717,16 @@ protected void serviceStop() throws Exception {
if (this.fedRenameScheduler != null) {
fedRenameScheduler.shutDown();
}
for (ThreadPoolExecutor executor : asyncRouterHandlerExecutors.values()) {
executor.shutdownNow();
}
asyncRouterHandlerExecutors.clear();
if (routerDefaultAsyncHandlerExecutor != null) {
routerDefaultAsyncHandlerExecutor.shutdownNow();
}
if (routerAsyncResponderExecutor != null) {
routerAsyncResponderExecutor.shutdownNow();
}
super.serviceStop();
}

Expand Down Expand Up @@ -2490,14 +2534,6 @@ public boolean isAsync() {
return this.enableAsync;
}

public Map<String, ExecutorService> getAsyncRouterHandlerExecutors() {
return asyncRouterHandlerExecutors;
}

public ExecutorService getRouterAsyncHandlerDefaultExecutor() {
return routerDefaultAsyncHandlerExecutor;
}

private static class AsyncThreadFactory implements ThreadFactory {
private final String namePrefix;
private final AtomicInteger threadNumber = new AtomicInteger(1);
Expand Down
14 changes: 10 additions & 4 deletions ...main/java/org/apache/hadoop/hdfs/server/federation/router/async/RouterAsyncRpcClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;

import static org.apache.hadoop.hdfs.server.federation.router.async.utils.Async.warpCompletionException;
import static org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil.asyncApply;
Expand Down Expand Up @@ -176,10 +177,11 @@ public Object invokeMethod(
// transfer threadLocalContext to worker threads of executor.
ThreadLocalContext threadLocalContext = new ThreadLocalContext();
asyncComplete(null);
// Returns a CompletableFuture with RejectedExecutionException if nsExecutor is full.
asyncApplyUseExecutor((AsyncApplyFunction<Object, Object>) o -> {
if (LOG.isDebugEnabled()) {
LOG.debug("Async invoke method : {}, {}, {}, {}", method.getName(), useObserver,
namenodes.toString(), params);
LOG.debug("Async invoke method : {}, {}, {}, {}", method.getName(), useObserver, namenodes,
params);
}
threadLocalContext.transfer();
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
Expand All @@ -190,8 +192,12 @@ public Object invokeMethod(
releasePermit(nsid, ugi, method, controller);
return object;
});
}, router.getRpcServer().getAsyncRouterHandlerExecutors().getOrDefault(nsid,
router.getRpcServer().getRouterAsyncHandlerDefaultExecutor()));
}, router.getRpcServer().getAsyncExecutorForNamespace(nsid));

// Catch the RejectedExecutionException and convert it to StandbyException
asyncCatch((ret, e) -> {
throw new StandbyException("Namespace '" + nsid + "' async handler is busy.");
}, RejectedExecutionException.class);
return null;
}

Expand Down
10 changes: 10 additions & 0 deletions hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/resources/hdfs-rbf-default.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,16 @@
</description>
</property>

<property>
<name>dfs.federation.router.async.rpc.queue.size</name>
<value>1000</value>
<description>
The bounded queue capacity for each per-nameservice async handler thread
pool. When the queue is full, new requests are rejected with a
StandbyException to prevent unbounded memory growth. Must be at least 1.
</description>
</property>

<property>
<name>dfs.federation.router.async.rpc.responder.count</name>
<value>10</value>
Expand Down
Loading
Loading