szetszwo commented on code in PR #11218:
URL: https://github.com/apache/ozone/pull/11218#discussion_r4128406062
##########
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);
Review Comment:
Could they be updated without copying the map?
```java
Map<String, SCMProxyInfo> oldProxyInfoMap = scmProxyInfoMap;
scmProxyInfoMap = newConfig.proxyInfoMap;
```
##########
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:
Why not just stop the stale proxies in the loop above?
##########
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);
Review Comment:
- Call indexOf(..) first to avoid calling contains(..).
- In this else case, does it need to update currentProxySCMNodeId?
--
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]
