fix the problem

Author: wanghangyu817

docs/docs/high_availability.md

@@ -63,14 +63,14 @@ Prepare two or more servers to serve as the maxwell host server and a zookeeper
 Example Running Scripts:
 
 ```
-    bin/maxwell --log_level='INFO' --user='<user>' --password='<passwd>' --host='<host>' --producer=stdout --client_id='<client_id>' --ha=true --zookeeper_server ='<host1:port>,<host2:port>,<host3:port>' --ha_modality='zookeeper'
+    bin/maxwell --log_level='INFO' --user='<user>' --password='<passwd>' --host='<host>' --producer=stdout --client_id='<client_id>' --ha='zookeeper' --zookeeper_server ='<host1:port>,<host2:port>,<host3:port>'
 ```
 
 Run the preceding command on each maxwell host.
 
 Get which host is the leader script Example:
 ```
-    bin/maxwell-leaders --ha=true --ha_modality='zookeeper' --zookeeper_server ='<host1:port>,<host2:port>,<host3:port>' --client_id='<client_id>'
+    bin/maxwell-leaders --ha='zookeeper' --zookeeper_server ='<host1:port>,<host2:port>,<host3:port>' --client_id='<client_id>'
 ```
 You can get:
 ```
@@ -80,9 +80,9 @@ You can get:
 ## Getting deeper
 If a timeout error occurs between the maxwell host and the zookeeper cluster or the connection is abnormal due to network instability, you can set the following parameters:
 ```
---session_time_out_ms=<session timeout duration> 
---connection_time_out_ms=<internal default wait time for the client to establish a connection with the zk> 
---max_retries=<number of retries>
---base_sleep_time_ms=<retry time interval>
+--zookeeper_session_timeout_ms=<session timeout duration> 
+--zookeeper_connection_timeout_ms=<internal default wait time for the client to establish a connection with the zk> 
+--zookeeper_max_retries=<number of retries>
+--zookeeper_retry_wait_ms=<retry time interval>
 ```
 

src/main/java/com/zendesk/maxwell/Maxwell.java

@@ -332,19 +332,18 @@ public void run() {
 
 			LOGGER.info("Starting Maxwell. maxMemory: " + Runtime.getRuntime().maxMemory() + " bufferMemoryUsage: " + config.bufferMemoryUsage);
 
-			if ( config.haMode ) {
-
-				if ( "jgroups-raft".equals(config.haModality)) {
-					new MaxwellHA(maxwell, config.jgroupsConf, config.raftMemberID, config.clientID).startHA();
-				} else if ("zookeeper".equals(config.haModality)){
+			if ( null != config.haMode){
+				if ( "jgroups-raft".equals(config.haMode)){
+					new MaxwellHA(maxwell, config.jgroupsConf, config.raftMemberID, config.clientID).startHAJGroups();
+				} else if ( "zookeeper".equals(config.haMode)){
 					if( StringUtils.isBlank(config.zookeeperServer)){
 						throw new Exception("In high availability mode 'zookeeperServer' does not allow Null. --zookeeper_server = " + config.zookeeperServer);
 					}
-					new MaxwellHA(maxwell, config.zookeeperServer, config.sessionTimeoutMs, config.connectionTimeoutMs, config.maxRetries, config.baseSleepTimeMs, config.clientID).startHA(config.haModality);
-				}else {
-					throw new Exception("Currently only jgroups-raft and zookeeper are supported. Check that the value of '--ha_modality' is one of them");
+					new MaxwellHA(maxwell, config.zookeeperServer, config.zookeeperSessionTimeoutMs, config.zookeeperConnectionTimeoutMs, config.zookeeperMaxRetries, config.zookeeperRetryWaitMs, config.clientID).startHAZookeeper();
+				} else {
+					throw new Exception("The value of ha is not in (jgroups-raft,zookeeper). ha = " + config.haMode);
 				}
-			} else {
+			} else{
 				maxwell.start();
 			}
 		} catch ( SQLException e ) {

src/main/java/com/zendesk/maxwell/MaxwellConfig.java

@@ -611,7 +611,7 @@ public class MaxwellConfig extends AbstractConfig {
 	/**
 	 * Enable high available support (via jgroups-raft or zookeeper)
 	 */
-	public boolean haMode;
+	public String haMode;
 
 	/**
 	 * Path to raft.xml file that configures high availability support
@@ -637,27 +637,23 @@ public class MaxwellConfig extends AbstractConfig {
 	/**
 	 * session time
 	 */
-	public int sessionTimeoutMs;
+	public int zookeeperSessionTimeoutMs;
 
 	/**
 	 * connection time
 	 */
-	public int connectionTimeoutMs;
+	public int zookeeperConnectionTimeoutMs;
 
 	/**
 	 * maxRetries
 	 */
-	public int maxRetries;
+	public int zookeeperMaxRetries;
 
 	/**
-	 * baseSleepTimeMs
+	 * retryWaitMs
 	 */
-	public int baseSleepTimeMs;
+	public int zookeeperRetryWaitMs;
 
-	/**
-	 * jgroups-raft or zookeeper
-	 */
-	public String haModality;
 
 	/**
 	 * Build a default configuration object.
@@ -772,23 +768,21 @@ protected MaxwellOptionParser buildOptionParser() {
 		parser.separator();
 
 		parser.accepts( "ha", "enable high-availability mode via jgroups-raft or zookeeper" )
-				.withOptionalArg().ofType(Boolean.class);
+				.withOptionalArg();
 		parser.accepts( "jgroups_config", "location of jgroups xml configuration file" )
 				.withRequiredArg();
 		parser.accepts( "raft_member_id", "raft memberID.  (may also be specified in raft.xml)" )
 				.withRequiredArg();
 		parser.accepts("zookeeper_server","enable maxwell High Availability using zookeeper")
 				.withRequiredArg();
-		parser.accepts("session_time_out_ms","session timeout duration (maxwellHA on zk)")
+		parser.accepts("zookeeper_session_timeout_ms","session timeout duration (maxwellHA on zk)")
 				.withRequiredArg().ofType(Integer.class);
-		parser.accepts("connection_time_out_ms","connection timeout duration (maxwellHA on zk)")
+		parser.accepts("zookeeper_connection_timeout_ms","connection timeout duration (maxwellHA on zk)")
 				.withRequiredArg().ofType(Integer.class);
-		parser.accepts("max_retries","maximum retry (maxwellHA on zk)")
+		parser.accepts("zookeeper_max_retries","maximum retry (maxwellHA on zk)")
 				.withRequiredArg().ofType(Integer.class);
-		parser.accepts("base_sleep_time_ms","initial retry wait time (maxwellHA on zk)")
+		parser.accepts("zookeeper_retry_wait_ms","initial retry wait time (maxwellHA on zk)")
 				.withRequiredArg().ofType(Integer.class);
-		parser.accepts("ha_modality","high availability mode : jgroups-raft or zookeeper")
-				.withRequiredArg();
 
 		parser.separator();
 
@@ -1248,17 +1242,16 @@ private void setup(OptionSet options, Properties properties) {
 
 		setupEncryptionOptions(options, properties);
 
-		this.haMode = fetchBooleanOption("ha", options, properties, false);
+		this.haMode = fetchStringOption("ha", options, properties, null);
 		this.jgroupsConf = fetchStringOption("jgroups_config", options, properties, "raft.xml");
 		this.raftMemberID = fetchStringOption("raft_member_id", options, properties, null);
 		this.replicationReconnectionRetries = fetchIntegerOption("replication_reconnection_retries", options, properties, 1);
 
 		this.zookeeperServer = fetchStringOption("zookeeper_server", options, properties, null);
-		this.sessionTimeoutMs = fetchIntegerOption("session_time_out_ms", options, properties, 6000);
-		this.connectionTimeoutMs = fetchIntegerOption("connection_time_out_ms", options, properties, 6000);
-		this.maxRetries = fetchIntegerOption("max_retries", options, properties, 3);
-		this.baseSleepTimeMs = fetchIntegerOption("base_sleep_time_ms", options, properties, 1000);
-		this.haModality = fetchStringOption("ha_modality", options, properties, "jgroups-raft");
+		this.zookeeperSessionTimeoutMs = fetchIntegerOption("zookeeper_session_timeout_ms", options, properties, 6000);
+		this.zookeeperConnectionTimeoutMs = fetchIntegerOption("zookeeper_connection_timeout_ms", options, properties, 6000);
+		this.zookeeperMaxRetries = fetchIntegerOption("zookeeper_max_retries", options, properties, 3);
+		this.zookeeperRetryWaitMs = fetchIntegerOption("zookeeper_retry_wait_ms", options, properties, 1000);
 
 		this.binlogEventQueueSize = fetchIntegerOption("binlog_event_queue_size", options, properties, BinlogConnectorReplicator.BINLOG_QUEUE_SIZE);
 	}

src/main/java/com/zendesk/maxwell/MaxwellHA.java

@@ -79,7 +79,7 @@ private void run() {
 	 * Does not return.
 	 * @throws Exception if there's any issues
 	 */
-	public void startHA() throws Exception {
+	public void startHAJGroups() throws Exception {
 		JChannel ch=new JChannel(jgroupsConf);
 		RaftHandle handle=new RaftHandle(ch, null);
 		if ( raftMemberID != null )
@@ -112,10 +112,9 @@ public void startHA() throws Exception {
 
 	/**
 	 * indicates that Ha is started in zookeeper mode
-	 * @param flag the overloading of the method facilitates the subsequent expansion of multiple modes
 	 * @throws Exception
 	 */
-	public void startHA(String flag) throws Exception {
+	public void startHAZookeeper() throws Exception {
 		String electPath = "/" + clientID + "/services";
 		String masterPath = "/" + clientID + "/leader";
 		CuratorUtils cu = new CuratorUtils();
@@ -131,17 +130,26 @@ public void startHA(String flag) throws Exception {
 		CuratorFramework client = cu.getClient();
 		LeaderLatch leader = new LeaderLatch(client, cu.getElectPath());
 		leader.start();
+		LOGGER.info("this node is participating in the election of the leader ....");
 		leader.addListener(new LeaderLatchListener() {
 			@Override
 			public void isLeader() {
-				cu.register();
+				try {
+					cu.register();
+				} catch (Exception e) {
+					e.printStackTrace();
+					LOGGER.error("The node registration is abnormal, check whether the maxwell host communicates properly with the zookeeper network");
+					cu.stop();
+					System.exit(1);
+				}
+				LOGGER.info("node is current leader, starting Maxwell....");
 				run();
-				LOGGER.debug("starting maxwell");
+				cu.stop();
 			}
 
 			@Override
 			public void notLeader() {
-
+				//LeaderLatch.CloseMode.SILENT mode will not invoke this method
 			}
 		});
 

src/main/java/com/zendesk/maxwell/util/CuratorUtils.java

@@ -5,17 +5,13 @@
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.zookeeper.CreateMode;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.List;
 
 public class CuratorUtils {
 
-	static final Logger LOGGER = LoggerFactory.getLogger(CuratorUtils.class);
-
 	private CuratorFramework client;
 	private String zookeeperServer;
 	private int sessionTimeoutMs;
@@ -104,27 +100,19 @@ public CuratorFramework getClient() {
 		return client;
 	}
 
-	public void register() {
-		try {
-			String rootPath = masterPath;
-			String hostAddress = InetAddress.getLocalHost().getHostAddress();
-			client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(rootPath + "/" + hostAddress);
-		} catch (Exception e) {
-			LOGGER.error("register exception", e);
-		}
+	public void register() throws Exception {
+		String rootPath = masterPath;
+		String hostAddress = InetAddress.getLocalHost().getHostAddress();
+		client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(rootPath + "/" + hostAddress);
 	}
 
-	public List<String> getChildren(String path) {
+	public List<String> getChildren(String path) throws Exception {
 		List<String> childrenList = new ArrayList<>();
-		try {
-			childrenList = client.getChildren().forPath(path);
-		} catch (Exception e) {
-			LOGGER.error("There was an error getting the child nodes", e);
-		}
+		childrenList = client.getChildren().forPath(path);
 		return childrenList;
 	}
 
-	public List<String> getInstances() {
+	public List<String> getInstances() throws Exception {
 		return getChildren(masterPath);
 	}
 }

src/main/java/com/zendesk/maxwell/util/MaxwellLeaders.java

@@ -23,28 +23,35 @@ public static void main(String[] args) {
 			Logging.setLevel("INFO");
 		}
 
-		if( config.haMode && "zookeeper".equals(config.haModality)){
+		if( "zookeeper".equals(config.haMode)){
 			CuratorUtils cu = new CuratorUtils();
 			cu.setZookeeperServer(config.zookeeperServer);
-			cu.setSessionTimeoutMs(config.sessionTimeoutMs);
-			cu.setConnectionTimeoutMs(config.connectionTimeoutMs);
-			cu.setMaxRetries(config.maxRetries);
-			cu.setBaseSleepTimeMs(config.baseSleepTimeMs);
+			cu.setSessionTimeoutMs(config.zookeeperSessionTimeoutMs);
+			cu.setConnectionTimeoutMs(config.zookeeperConnectionTimeoutMs);
+			cu.setMaxRetries(config.zookeeperMaxRetries);
+			cu.setBaseSleepTimeMs(config.zookeeperRetryWaitMs);
 			cu.setClientId(config.clientID);
 			String electPath = "/" + config.clientID + "/services";
 			String masterPath = "/" + config.clientID + "/leader";
 			cu.setElectPath(electPath);
 			cu.setMasterPath(masterPath);
 			cu.init();
-			List<String> instances = cu.getInstances();
+			List<String> instances = null;
+			try {
+				instances = cu.getInstances();
+			} catch (Exception e) {
+				e.printStackTrace();
+				LOGGER.error("The path does not exist or is empty. Please check whether the clientID is correct. clientID = " + config.clientID);
+				System.exit(1);
+			}
 
 			if(0 == instances.size()){
 				LOGGER.info("Maxwell is not a high availability mode Or maxwell is not started");
 			}else {
 				LOGGER.info("clientID:"+config.clientID + ":leaders now are -> {}",instances.get(0));
 			}
 		}else {
-			LOGGER.info("Maxwell is not a high availability mode");
+			LOGGER.error("make sure ha = 'zookeeper'. ha = " + config.haMode);
 		}
 	}
 }