kokonguyen191 commented on code in PR #8548:
URL: https://github.com/apache/hadoop/pull/8548#discussion_r3435326285
##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java:
##########
@@ -510,44 +515,75 @@ public RouterRpcServer(Configuration conf, Router router,
public void initAsyncThreadPools(Configuration configuration) {
Set<String> allConfiguredNS =
FederationUtil.getAllConfiguredNS(configuration);
allConfiguredNS.add(CONCURRENT_NS);
- Map<String, Integer> nsAsyncActiveHandlerCount =
parseNsAsyncHandlerCount(configuration);
- initAsyncHandlerThreadPools(configuration, allConfiguredNS,
nsAsyncActiveHandlerCount);
+ Map<String, Integer> nsAsyncActiveHandlerCount =
parseNsAsyncHandlerCount(configuration,
+ DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY);
+ initAsyncHandlerThreadPools(configuration, allConfiguredNS, false,
+ DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT,
+ nsAsyncActiveHandlerCount, asyncRouterHandlerExecutors);
+
+ Map<String, Integer> nsAsyncObserverHandlerCount =
parseNsAsyncHandlerCount(configuration,
+ DFS_ROUTER_ASYNC_RPC_NS_OBSERVER_HANDLER_COUNT_KEY);
+ initAsyncHandlerThreadPools(configuration, allConfiguredNS, true,
+ DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY,
+ DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_DEFAULT,
nsAsyncObserverHandlerCount,
+ asyncRouterOBHandlerExecutors);
+
initAsyncResponderThreadPools(configuration);
}
- private void initAsyncHandlerThreadPools(Configuration configuration,
- Set<String> allConfiguredNS, Map<String, Integer> nsAsyncHandlerCount) {
- LOG.info("Initializing asynchronous handler thread pools");
+ /**
+ * Initializes async handler executors for each configured namespace.
+ *
+ * @param configuration router configuration
+ * @param allConfiguredNS set of all configured namespace IDs
+ * @param useObserver true if these handlers serve observer namenodes
+ * @param handlerCountKey key for the default handler count
+ * @param handlerCountDefault default handler count
+ * @param nsAsyncHandlerCount per-namespace handler counts
+ * @param executors executor map to populate
+ */
+ private void initAsyncHandlerThreadPools(Configuration configuration,
Set<String> allConfiguredNS,
+ boolean useObserver, String handlerCountKey, int handlerCountDefault,
+ Map<String, Integer> nsAsyncHandlerCount, Map<String,
ThreadPoolExecutor> executors) {
+ String namenodeTypeForLogging = useObserver ? "Observer Namenode" :
"Active Namenode";
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);
+ int asyncHandlerCountDefault = configuration.getInt(handlerCountKey,
handlerCountDefault);
+ // Skip separate observer handlers and let them share with active if not
configured
+ if (asyncHandlerCountDefault < 1 && nsAsyncHandlerCount.isEmpty() &&
useObserver) {
+ LOG.info("Async observer handlers are not configured, skipping...");
+ return;
+ } else {
+ useSeparateAsyncRouterOBHandlerExecutors = true;
+ }
Review Comment:
oof, good catch. Fixed and added unit test
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
