support zkAddress updated on flight

core/src/main/java/com/ctrip/xpipe/config/ConfigKeyListener.java

@@ -0,0 +1,11 @@
+package com.ctrip.xpipe.config;
+
+/**
+ * @author lishanglin
+ * date 2024/3/8
+ */
+public interface ConfigKeyListener {
+
+    void onChange(String key, String val);
+
+}

core/src/main/java/com/ctrip/xpipe/config/ZkConfig.java

@@ -7,6 +7,9 @@
  */
 public interface ZkConfig {
 
+    String KEY_ZK_ADDRESS  = "zk.address";
+    String KEY_ZK_NAMESPACE  = "zk.namespace";
+
     String getZkConnectionString();
 
     String getZkNameSpace();

core/src/main/java/com/ctrip/xpipe/spring/AbstractProfile.java

@@ -22,16 +22,10 @@
 
 
 	protected ZkClient getZkClient(String zkNameSpace, String zkAddress){
-
-		DefaultZkClient zkClient = new DefaultZkClient();
 
-		DefaultZkConfig zkConfig = new DefaultZkConfig();
+		DefaultZkConfig zkConfig = new DefaultZkConfig(zkAddress);
 		zkConfig.setZkNameSpace(zkNameSpace);
-
-		zkClient.setZkConfig(zkConfig);
-		zkClient.setZkAddress(zkAddress);
-		return zkClient;
-
+		return new DefaultZkClient(zkConfig);
 	}
 
 }

core/src/main/java/com/ctrip/xpipe/zk/ZkClient.java

@@ -1,13 +1,14 @@
 package com.ctrip.xpipe.zk;
 
+import com.ctrip.xpipe.config.ConfigKeyListener;
 import org.apache.curator.framework.CuratorFramework;
 
 /**
  * @author marsqing
  *
  *         Jun 16, 2016 12:05:46 PM
  */
-public interface ZkClient{
+public interface ZkClient extends ConfigKeyListener {
 
 	CuratorFramework get();
 

core/src/main/java/com/ctrip/xpipe/zk/ZkConfig.java

@@ -24,7 +24,11 @@ public interface ZkConfig {
 	int getZkSessionTimeoutMillis();
 
 	int waitForZkConnectedMillis();
-
-	CuratorFramework create(String address) throws InterruptedException;
+
+	String getZkAddress();
+
+	void updateZkAddress(String address);
+
+	CuratorFramework create() throws InterruptedException;
 
 }

core/src/main/java/com/ctrip/xpipe/zk/impl/DefaultZkClient.java

@@ -16,11 +16,13 @@
  */
 public class DefaultZkClient extends AbstractLifecycle implements ZkClient, TopElement, Lifecycle {
 
-	private ZkConfig zkConfig = new DefaultZkConfig();
+	private ZkConfig zkConfig;
 
 	private CuratorFramework client;
-
-	private String address;
+
+	public DefaultZkClient(ZkConfig zkConfig) {
+		this.zkConfig = zkConfig;
+	}
 
 	protected void doInitialize() throws Exception {
 
@@ -29,15 +31,22 @@
 	@Override
 	protected void doStart() throws Exception {
 
-		logger.info("[doStart]{}", address);
-		client= zkConfig.create(address);
+		logger.info("[doStart]{}", this.zkConfig.getZkAddress());
+		client= zkConfig.create();
 	}
 
 
 	@Override
 	protected void doStop() throws Exception {
 		client.close();
 	}
+
+	@Override
+	public void onChange(String key, String val) {
+		if (key.equalsIgnoreCase(com.ctrip.xpipe.config.ZkConfig.KEY_ZK_ADDRESS)) {
+			this.zkConfig.updateZkAddress(val);
+		}
+	}
 
 	@Override
 	public CuratorFramework get() {
@@ -46,16 +55,12 @@
 
 	@Override
 	public void setZkAddress(String address) {
-		this.address = address;
+		this.zkConfig.updateZkAddress(address);
 	}
 
 	@Override
 	public String getZkAddress(){
-		return this.address;
-	}
-
-	public void setZkConfig(ZkConfig zkConfig) {
-		this.zkConfig = zkConfig;
+		return this.zkConfig.getZkAddress();
 	}
 
 	@Override

core/src/main/java/com/ctrip/xpipe/zk/impl/DefaultZkConfig.java

@@ -3,6 +3,8 @@
 import com.ctrip.xpipe.api.codec.Codec;
 import com.ctrip.xpipe.utils.XpipeThreadFactory;
 import com.ctrip.xpipe.zk.ZkConfig;
+import org.apache.curator.ensemble.EnsembleProvider;
+import org.apache.curator.ensemble.fixed.FixedEnsembleProvider;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
@@ -27,7 +29,23 @@ public class DefaultZkConfig implements ZkConfig{
 	private int zkConnectionTimeoutMillis = Integer.parseInt(System.getProperty("ZK.CONN.TIMEOUT", "3000"));
 	private int zkRetries = 3;
 	private String zkNameSpace = System.getProperty(KEY_ZK_NAMESPACE, DEFAULT_ZK_NAMESPACE);
-
+	private EnsembleProvider ensembleProvider;
+
+	public DefaultZkConfig(String address) {
+		this.ensembleProvider = new FixedEnsembleProvider(address, true);
+	}
+
+	@Override
+	public void updateZkAddress(String address) {
+		logger.info("[updateZkAddress] {} -> {}", this.ensembleProvider.getConnectionString(), address);
+		this.ensembleProvider.setConnectionString(address);
+	}
+
+	@Override
+	public String getZkAddress() {
+		return this.ensembleProvider.getConnectionString();
+	}
+
 	@Override
 	public int getZkConnectionTimeoutMillis() {
 		return zkConnectionTimeoutMillis;
@@ -80,18 +98,18 @@ public int waitForZkConnectedMillis() {
 	}
 
 	@Override
-	public CuratorFramework create(String address) throws InterruptedException {
+	public CuratorFramework create() throws InterruptedException {
 
 		Builder builder = CuratorFrameworkFactory.builder();
 		builder.connectionTimeoutMs(getZkConnectionTimeoutMillis());
-		builder.connectString(address);
+		builder.ensembleProvider(this.ensembleProvider);
 		builder.maxCloseWaitMs(getZkCloseWaitMillis());
 		builder.namespace(getZkNamespace());
 		builder.retryPolicy(new RetryNTimes(getZkRetries(), getSleepMsBetweenRetries()));
 		builder.sessionTimeoutMs(getZkSessionTimeoutMillis());
-		builder.threadFactory(XpipeThreadFactory.create("Xpipe-ZK-" + address, true));
+		builder.threadFactory(XpipeThreadFactory.create("Xpipe-ZK-" + this.ensembleProvider.getConnectionString(), true));
 
-		logger.info("[create]{}, {}", Codec.DEFAULT.encode(this), address);
+		logger.info("[create]{}, {}", Codec.DEFAULT.encode(this), this.ensembleProvider.getConnectionString());
 		CuratorFramework curatorFramework = builder.build();
 		curatorFramework.start();
 		curatorFramework.blockUntilConnected(waitForZkConnectedMillis(), TimeUnit.MILLISECONDS);

core/src/main/java/com/ctrip/xpipe/zk/impl/SpringZkClient.java

@@ -16,18 +16,22 @@
 public class SpringZkClient implements ZkClient{
 
     private ZkConfig zkConfig;
-    private String zkAddress;
     private CuratorFramework curatorFramework;
 
-    public SpringZkClient(ZkConfig zkConfig, String zkAddress){
+    public SpringZkClient(ZkConfig zkConfig){
         this.zkConfig = zkConfig;
-        this.zkAddress = zkAddress;
     }
 
+    @Override
+    public void onChange(String key, String val) {
+        if (key.equalsIgnoreCase(com.ctrip.xpipe.config.ZkConfig.KEY_ZK_ADDRESS)) {
+            this.zkConfig.updateZkAddress(val);
+        }
+    }
 
     @PostConstruct
     public void postContruct() throws InterruptedException {
-        curatorFramework = zkConfig.create(zkAddress);
+        curatorFramework = zkConfig.create();
     }
 
     @Override
@@ -37,12 +41,12 @@
 
     @Override
     public void setZkAddress(String zkAddress) {
-        this.zkAddress = zkAddress;
+        this.zkConfig.updateZkAddress(zkAddress);
     }
 
     @Override
     public String getZkAddress() {
-        return zkAddress;
+        return this.zkConfig.getZkAddress();
     }
 
     @PreDestroy

core/src/main/java/com/ctrip/xpipe/zk/impl/TestZkClient.java

@@ -16,11 +16,11 @@
 
 	private volatile CuratorFramework client;
 
-	private ZkConfig zkConfig = new DefaultZkConfig();
-
 	public static final String ZK_ADDRESS_KEY = "zkAddress";
 
 	private String address = System.getProperty(ZK_ADDRESS_KEY, "127.0.0.1:2181");
+
+	private ZkConfig zkConfig = new DefaultZkConfig(address);
 
 	protected void doInitialize() throws Exception {
 
@@ -38,6 +38,13 @@
 			client = null;
 		}
 	}
+
+	@Override
+	public void onChange(String key, String val) {
+		if (key.equalsIgnoreCase(com.ctrip.xpipe.config.ZkConfig.KEY_ZK_ADDRESS)) {
+			this.zkConfig.updateZkAddress(val);
+		}
+	}
 
 	@Override
 	public synchronized CuratorFramework get() {
@@ -52,7 +59,7 @@
 		}
 
 		try {
-			client = zkConfig.create(address);
+			client = zkConfig.create();
 			return client;
 		} catch (InterruptedException e) {
 			logger.error("[get]", e);
@@ -62,12 +69,12 @@
 
 	@Override
 	public void setZkAddress(String address) {
-		this.address = address;
+		this.zkConfig.updateZkAddress(address);
 	}
 
 	@Override
 	public String getZkAddress(){
-		return this.address;
+		return this.zkConfig.getZkAddress();
 	}
 
 	public void setClient(CuratorFramework client) {

core/src/test/java/com/ctrip/xpipe/cluster/DefaultLeaderElectorTest.java

@@ -24,8 +24,8 @@ public class DefaultLeaderElectorTest extends AbstractTest{
 	public void beforeDefaultLeaderElectorTest() throws InterruptedException{
 		zkTestServer = startRandomZk();
 
-		ZkConfig config = new DefaultZkConfig();
-		client = config.create(String.format("localhost:%d", zkTestServer.getZkPort()));
+		ZkConfig config = new DefaultZkConfig(String.format("localhost:%d", zkTestServer.getZkPort()));
+		client = config.create();
 
 	}
 

core/src/test/java/com/ctrip/xpipe/zk/impl/DefaultZkConfigTest.java

@@ -17,27 +17,27 @@ public class DefaultZkConfigTest {
 
     @Test
     public void testGetZkConnectionTimeoutMillisDefaultValue() {
-        zkConfig = new DefaultZkConfig();
+        zkConfig = new DefaultZkConfig("localhost:2181");
         Assert.assertEquals(3000, zkConfig.getZkConnectionTimeoutMillis());
     }
 
     @Test
     public void testGetZkConnectionTimeoutMillis() {
         System.setProperty("ZK.CONN.TIMEOUT", "1000");
-        zkConfig = new DefaultZkConfig();
+        zkConfig = new DefaultZkConfig("localhost:2181");
         Assert.assertEquals(1000, zkConfig.getZkConnectionTimeoutMillis());
     }
 
     @Test
     public void testGetZkSessionTimeoutMillisDefaultValue() {
-        zkConfig = new DefaultZkConfig();
+        zkConfig = new DefaultZkConfig("localhost:2181");
         Assert.assertEquals(5000, zkConfig.getZkSessionTimeoutMillis());
     }
 
     @Test
     public void testGetZkSessionTimeoutMillis() {
         System.setProperty("ZK.SESSION.TIMEOUT", "1000");
-        zkConfig = new DefaultZkConfig();
+        zkConfig = new DefaultZkConfig("localhost:2181");
         Assert.assertEquals(1000, zkConfig.getZkSessionTimeoutMillis());
     }
 }

core/src/test/java/com/ctrip/xpipe/zk/usage/AbstractZkUsageTest.java

@@ -19,8 +19,8 @@ public class AbstractZkUsageTest extends AbstractTest{
 	@Before
 	public void beforeAbstractZkUsageTest() throws InterruptedException{
 
-		ZkConfig zkConfig = new DefaultZkConfig();
-		client = zkConfig.create(zkAddress);
+		ZkConfig zkConfig = new DefaultZkConfig(zkAddress);
+		client = zkConfig.create();
 	}
 
 	public CuratorFramework getClient() {

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/config/ConsoleConfig.java

@@ -4,7 +4,6 @@
 import com.ctrip.xpipe.redis.checker.config.CheckerConfig;
 import com.ctrip.xpipe.redis.checker.healthcheck.actions.interaction.DcClusterDelayMarkDown;
 import com.ctrip.xpipe.redis.console.config.model.BeaconOrgRoute;
-import com.ctrip.xpipe.redis.console.model.KeeperContainerOverloadStandardModel;
 import com.ctrip.xpipe.redis.console.util.HickwallMetricInfo;
 import com.ctrip.xpipe.redis.core.config.CoreConfig;
 import com.ctrip.xpipe.redis.core.meta.QuorumConfig;

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/config/impl/DefaultConsoleConfig.java

@@ -7,7 +7,6 @@
 import com.ctrip.xpipe.redis.checker.healthcheck.actions.interaction.DcClusterDelayMarkDown;
 import com.ctrip.xpipe.redis.console.config.ConsoleConfig;
 import com.ctrip.xpipe.redis.console.config.model.BeaconOrgRoute;
-import com.ctrip.xpipe.redis.console.model.KeeperContainerOverloadStandardModel;
 import com.ctrip.xpipe.redis.console.util.HickwallMetricInfo;
 import com.ctrip.xpipe.redis.core.config.AbstractCoreConfig;
 import com.ctrip.xpipe.redis.core.meta.QuorumConfig;

redis/redis-console/src/main/java/com/ctrip/xpipe/redis/console/spring/Production.java

@@ -21,17 +21,17 @@
 
     @Bean
     public ZkClient getZkClient(ConsoleConfig consoleConfig) {
-        return getZkClient(consoleConfig.getZkNameSpace(), consoleConfig.getZkConnectionString());
+        ZkClient zkClient = getZkClient(consoleConfig.getZkNameSpace(), consoleConfig.getZkConnectionString());
+        consoleConfig.addListener(zkClient);
+        return zkClient;
     }
 
     @Override
     protected ZkClient getZkClient(String zkNameSpace, String zkAddress) {
 
-        DefaultZkConfig zkConfig = new DefaultZkConfig();
+        DefaultZkConfig zkConfig = new DefaultZkConfig(zkAddress);
         zkConfig.setZkNameSpace(zkNameSpace);
-
-        SpringZkClient springZkClient = new SpringZkClient(zkConfig, zkAddress);
-        return springZkClient;
+        return new SpringZkClient(zkConfig);
     }
 
     @Bean