hani-fouladgar commented on code in PR #11218:
URL: https://github.com/apache/ozone/pull/11218#discussion_r4138400885
##########
hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java:
##########
@@ -188,16 +199,87 @@ protected synchronized void loadConfigs() {
String scmServiceId = scmNodeInfo.getServiceId();
String scmNodeId = scmNodeInfo.getNodeId();
- scmNodeIds.add(scmNodeId);
+ newScmNodeIds.add(scmNodeId);
// Preserve the original config string so DNS can be re-resolved
// on connection failure when the SCM peer is rescheduled to a
// new IP (Kubernetes pod-IP-change recovery). See
// refreshProxyAddressIfChanged(String).
SCMProxyInfo scmProxyInfo = new SCMProxyInfo(scmServiceId, scmNodeId,
protocolAddr, protocolAddress);
- scmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ newScmProxyInfoMap.put(scmNodeId, scmProxyInfo);
+ }
+ }
+
+ return new ScmProxyConfig(newScmNodeIds, newScmProxyInfoMap);
+ }
+
+ /**
+ * Reloads SCM nodes and proxies from updated config without a restart.
+ * Stops proxies for removed/changed nodes; fails atomically if config is
invalid.
+ * In-flight calls on removed nodes persist until the next failover.
+ */
+ public void changeConfig() {
+ // Resolve DNS before locking to prevent slow lookup from blocking callers.
+ ScmProxyConfig newConfig = buildConfigs();
+
+ Map<String, ProxyInfo<T>> staleProxies = new HashMap<>();
+ synchronized (this) {
+ Map<String, SCMProxyInfo> oldProxyInfoMap = new
HashMap<>(scmProxyInfoMap);
+ scmNodeIds = newConfig.nodeIds;
+ scmProxyInfoMap.clear();
+ scmProxyInfoMap.putAll(newConfig.proxyInfoMap);
+
+ // Re-sync proxy index to the new list, or fall back to first node if
removed.
+ if (!scmNodeIds.contains(currentProxySCMNodeId)) {
+ currentProxyIndex = 0;
+ currentProxySCMNodeId = scmNodeIds.get(currentProxyIndex);
+ } else {
+ currentProxyIndex = scmNodeIds.indexOf(currentProxySCMNodeId);
+ }
+
+ // Drop removed failover target to prevent NPE on next failover.
+ if (updatedLeaderNodeID != null
+ && !scmProxyInfoMap.containsKey(updatedLeaderNodeID)) {
+ updatedLeaderNodeID = null;
+ }
+
+ // Evict stale proxies under lock, but defer stopProxy until unlocked to
avoid blocking.
+ for (Map.Entry<String, SCMProxyInfo> entry : oldProxyInfoMap.entrySet())
{
+ String nodeId = entry.getKey();
+ SCMProxyInfo newInfo = scmProxyInfoMap.get(nodeId);
+ if (newInfo == null
+ || !newInfo.getAddress().equals(entry.getValue().getAddress())) {
+ ProxyInfo<T> staleProxy = scmProxies.remove(nodeId);
+ if (staleProxy != null && staleProxy.proxy != null) {
+ staleProxies.put(nodeId, staleProxy);
+ }
+ }
+ }
+ }
+
+ for (Map.Entry<String, ProxyInfo<T>> entry : staleProxies.entrySet()) {
+ try {
+ RPC.stopProxy(entry.getValue().proxy);
+ } catch (RuntimeException stopEx) {
+ getLogger().warn("Failed to stop stale proxy for SCM node {}",
+ entry.getKey(), stopEx);
}
Review Comment:
`RPC.stopProxy()` can block while the IPC client tears down its connections,
and doing that inside the `synchronized` block would stall every concurrent
`getProxy()` / `shouldRetry()` caller on the provider monitor. Same reason we
resolve DNS before locking. `refreshProxyAddressIfChanged()` already uses the
same evict-under-lock / stop-after-unlock shape.
--
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]
