High availabily via jgroups-raft

Author: osheroff

docs/docs/config.md

@@ -164,7 +164,14 @@ filter                         | STRING            | filter rules, eg `exclude:
 option                         | argument                            | description                                         | default
 -------------------------------|-------------------------------------| --------------------------------------------------- | -------
 encrypt                        | [ none | data | all ]     | encrypt mode: none = no encryption. "data": encrypt the `data` field only. `all`: encrypt entire maxwell message | none
-secret_key                     | STRING                              | specify the encryption key to be used               | null
+secret_key                     | string                              | specify the encryption key to be used               | null
+
+# high availability
+option                         | argument                            | description                                         | default
+-------------------------------|-------------------------------------| --------------------------------------------------- | -------
+ha                             |                                     | enable maxwell client HA                            |
+jgroups_config                 | string                              | location of xml configuration file for jGroups      | $PWD/raft.xml
+raft_member_id                 | string                              | uniquely identify this node within jgroups-raft cluster |
 
 # monitoring / metrics
 option                         | argument                            | description                                         | default

docs/docs/high_availability.md

@@ -0,0 +1,60 @@
+# High Availabilty
+
+As v1.29.0, Maxwell contains experiemental (alpha quality) client-side HA.
+Support for performing leader elections is done via
+[jgroups-raft](https://github.com/belaban/jgroups-raft).
+
+## Getting started
+
+First, copy `raft.xml.example` to `raft.xml`.  Edit to your liking, paying attention to:
+
+```
+    <raft.RAFT members="A,B,C" raft_id="${raft_id:undefined}"/>
+```
+
+Note that because we are using a RAFT-based leader election, we will have to spin up at least
+3 maxwell client nodes.
+
+
+Now start each of your HA maxwell nodes like this:
+
+```bash
+host1: $ bin/maxwell --ha --raft_member_id=A
+host2: $ bin/maxwell --ha --raft_member_id=B
+host3: $ bin/maxwell --ha --raft_member_id=C
+```
+
+if all goes well, the 3 nodes will communicate via multicast/UDP, elect one to
+be the cluster leader, and away you will go.  If one node is terminated or
+partitioned, a new election will be held to replace it.
+
+
+## Getting deeper
+
+More advanced (especially inter-DC) configurations may be implemented by
+editing `raft.xml`; you'll probably need to get the nodes to communicate with
+each other via TCP instead of UDP, and maybe tunnel through a firewall or two,
+good stuff like that.  It's of course out of scope for this document, so
+[check out the jgroups
+documentation](http://www.jgroups.org/manual/html/user-advanced.html), but if
+you come up with something good drop me a line.
+
+## Common problems
+
+Something I encountered right out of the gate was this:
+
+```
+12:37:53,135 WARN  UDP - failed to join /224.0.75.75:7500 on utun0: java.net.SocketException: Can't assign requested address
+```
+
+which can be worked around by forcing the JVM onto an ipv4 stack:
+
+```
+JAVA_OPTS="-Djava.net.preferIPv4Stack=true" bin/maxwell --ha --raft_member_id=B
+```
+
+
+
+
+
+

docs/mkdocs.yml

@@ -14,6 +14,7 @@ nav:
     - 'Data Format': 'dataformat.md'
     - 'Encryption': 'encryption.md'
     - 'Monitoring': 'monitoring.md'
+    - 'High Availability': 'high_availability.md'
     - 'Embedding': 'embedding.md'
     - 'Internals': 'schemas.md'
     - 'Compatibility': 'compat.md'

pom.xml

@@ -318,7 +318,7 @@
     <dependency>
       <groupId>org.jgroups</groupId>
       <artifactId>jgroups-raft</artifactId>
-      <version>0.5.3-SNAPSHOT</version>
+      <version>1.0.0.Final</version>
     </dependency>
   </dependencies>
 

raft.xml.example

@@ -0,0 +1,22 @@
+<?xml version='1.0' encoding='utf-8'?>
+<config xmlns="urn:org:jgroups"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/jgroups.xsd">
+    <UDP mcast_addr="228.8.8.8" mcast_port="${jgroups.udp.mcast_port:45588}"/>
+    <PING />
+    <MERGE3 />
+    <FD_SOCK/>
+    <FD_ALL/>
+    <VERIFY_SUSPECT timeout="1500"/>
+    <pbcast.NAKACK2 xmit_interval="500"/>
+    <UNICAST3 xmit_interval="500"/>
+    <pbcast.STABLE desired_avg_gossip="50000" max_bytes="4M"/>
+    <raft.NO_DUPES/>
+    <pbcast.GMS print_local_addr="true" join_timeout="2000"/>
+    <UFC max_credits="2M" min_threshold="0.4"/>
+    <MFC max_credits="2M" min_threshold="0.4"/>
+    <FRAG2 frag_size="60K"/>
+    <raft.ELECTION election_min_interval="500" election_max_interval="1000" heartbeat_interval="250"/>
+    <raft.RAFT members="A,B,C" raft_id="${raft_id:undefined}"/>
+    <raft.REDIRECT/>
+</config>

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

@@ -13,26 +13,14 @@
 import com.zendesk.maxwell.schema.*;
 import com.zendesk.maxwell.schema.columndef.ColumnDefCastException;
 import com.zendesk.maxwell.util.Logging;
-import org.jgroups.JChannel;
-import org.jgroups.protocols.raft.RaftLeaderException;
-import org.jgroups.protocols.raft.Role;
-import org.jgroups.protocols.raft.Settable;
-import org.jgroups.protocols.raft.StateMachine;
-import org.jgroups.raft.RaftHandle;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.net.URISyntaxException;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 public class Maxwell implements Runnable {
 	protected MaxwellConfig config;
@@ -58,13 +46,22 @@ public void run() {
 		}
 	}
 
+	public void restart() {
+		try {
+			this.context = new MaxwellContext(config);
+		} catch ( Exception e ) {
+			throw new RuntimeException(e);
+		}
+
+		run();
+	}
+
 	public void terminate() {
 		Thread terminationThread = this.context.terminate();
 		if (terminationThread != null) {
 			try {
 				terminationThread.join();
 			} catch (InterruptedException e) {
-				// ignore
 			}
 		}
 	}
@@ -182,91 +179,6 @@ private void logBanner(AbstractProducer producer, Position initialPosition) {
 	protected void onReplicatorStart() {}
 	protected void onReplicatorEnd() {}
 
-	private AtomicBoolean isLeader = new AtomicBoolean(false);
-
-	private void startHA() throws Exception {
-		JChannel ch=new JChannel(this.config.jgroupsConf);
-		StateMachine s= new StateMachine() {
-			@Override
-			public byte[] apply(byte[] bytes, int i, int i1) throws Exception {
-				return new byte[0];
-			}
-
-			@Override
-			public void readContentFrom(DataInput dataInput) throws Exception {
-
-			}
-
-			@Override
-			public void writeContentTo(DataOutput dataOutput) throws Exception {
-
-			}
-		};
-
-		RaftHandle handle=new RaftHandle(ch, s);
-		handle.raftId(this.config.raftMemberID);
-
-		handle.addRoleListener(role -> {
-			if(role == Role.Leader) {
-				LOGGER.info("won HA election, starting maxwell");
-				try {
-					isLeader.set(true);
-					this.start();
-				} catch ( Exception e ) {
-				} finally {
-					isLeader.set(false);
-				}
-
-			} else
-				LOGGER.info("lost HA election, current leader: " + handle.leader());
-			// stop singleton services
-		});
-
-		ch.connect(this.config.clientID);
-		LOGGER.info("enter HA group, current leader: " +  handle.leader());
-
-		new Thread(() -> {
-			int exceptionCount = 0;
-			while ( true ) {
-				byte[] b = new byte[] { (byte) 0x1 };
-				try {
-					handle.set(b, 0, 1, 5000, TimeUnit.MILLISECONDS);
-					LOGGER.debug("RAFT-heartbeat successful");
-					exceptionCount = 0;
-					if ( handle.isLeader() && !this.isLeader.get() ) {
-						LOGGER.info("RAFT-consensus available, restarting maxwell...");
-						try {
-							isLeader.set(true);
-							this.start();
-						} catch ( Exception e ) {
-						} finally {
-							isLeader.set(false);
-						}
-					}
-				} catch ( RaftLeaderException e ) {
-					LOGGER.warn("RAFT leader unavailable: " + e.getMessage());
-					exceptionCount++;
-				} catch ( TimeoutException e ) {
-					exceptionCount++;
-					LOGGER.warn("RAFT-heartbeat timed out.  Exception Count:" + exceptionCount);
-				} catch ( Exception e ) {
-					LOGGER.error("unexpected exception in RAFT-heartbeat", e);
-				}
-
-				if ( exceptionCount > 1 && isLeader.get() ) {
-					LOGGER.warn("RAFT consensus unavailable after " + exceptionCount + " tries, stopping maxwell");
-					this.context.shutdown(new AtomicBoolean(false));
-				}
-
-				try {
-					Thread.sleep(1000);
-				} catch (InterruptedException e) { }
-			}
-		}).start();
-
-
-		Thread.sleep(Long.MAX_VALUE);
-	}
 
 	private void start() throws Exception {
 		try {
@@ -368,7 +280,7 @@ public void run() {
 			});
 
 			if ( config.haMode ) {
-				maxwell.startHA();
+				new MaxwellHA(maxwell, config.jgroupsConf, config.raftMemberID, config.clientID).startHA();
 			} else {
 				maxwell.start();
 			}

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

@@ -86,9 +86,6 @@ public class MaxwellConfig extends AbstractConfig {
 	public MaxwellOutputConfig outputConfig;
 	public String log_level;
 
-	public MetricRegistry metricRegistry;
-	public HealthCheckRegistry healthCheckRegistry;
-
 	public int httpPort;
 	public String httpBindAddress;
 	public String httpPathPrefix;
@@ -162,8 +159,6 @@ public MaxwellConfig() { // argv is only null in tests
 		this.masterRecovery = false;
 		this.gtidMode = false;
 		this.bufferedProducerSize = 200;
-		this.metricRegistry = new MetricRegistry();
-		this.healthCheckRegistry = new HealthCheckRegistry();
 		this.outputConfig = new MaxwellOutputConfig();
 		setup(null, null); // setup defaults
 	}
@@ -224,9 +219,9 @@ protected MaxwellOptionParser buildOptionParser() {
 		parser.accepts( "max_schemas", "Maximum schemas to keep before triggering a compaction operation.  Default: unlimited" ).withRequiredArg();
 		parser.section("operation");
 
-		parser.accepts( "ha", "enable high-availability mode via jgroups-raft" ).withOptionalArg();
+		parser.accepts( "ha", "enable high-availability mode via jgroups-raft" );
 		parser.accepts( "jgroups_config", "location of jgroups xml configuration file" ).withRequiredArg();
-		parser.accepts( "raft_member_id", "raft memberID" ).withRequiredArg();
+		parser.accepts( "raft_member_id", "raft memberID.  (may also be specified in raft.xml)" ).withRequiredArg();
 
 		parser.accepts( "bootstrapper", "bootstrapper type: async|sync|none. default: async" ).withRequiredArg();
 		parser.accepts( "init_position", "initial binlog position, given as BINLOG_FILE:POSITION[:HEARTBEAT]" ).withRequiredArg();

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

@@ -1,5 +1,7 @@
 package com.zendesk.maxwell;
 
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.health.HealthCheckRegistry;
 import com.zendesk.maxwell.bootstrap.BootstrapController;
 import com.zendesk.maxwell.bootstrap.SynchronousBootstrapper;
 import com.zendesk.maxwell.filtering.Filter;
@@ -33,9 +35,9 @@
 public class MaxwellContext {
 	static final Logger LOGGER = LoggerFactory.getLogger(MaxwellContext.class);
 
-	private final ConnectionPool replicationConnectionPool;
-	private final ConnectionPool maxwellConnectionPool;
-	private final ConnectionPool rawMaxwellConnectionPool;
+	private ConnectionPool replicationConnectionPool;
+	private ConnectionPool maxwellConnectionPool;
+	private ConnectionPool rawMaxwellConnectionPool;
 	private final ConnectionPool schemaConnectionPool;
 	private final MaxwellConfig config;
 	private final MaxwellMetrics metrics;
@@ -58,11 +60,16 @@ public class MaxwellContext {
 	private BootstrapController bootstrapController;
 	private Thread bootstrapControllerThread;
 
+	public MetricRegistry metricRegistry;
+	public HealthCheckRegistry healthCheckRegistry;
+
 	public MaxwellContext(MaxwellConfig config) throws SQLException, URISyntaxException {
 		this.config = config;
 		this.config.validate();
 		this.taskManager = new TaskManager();
-		this.metrics = new MaxwellMetrics(config);
+
+		this.metricRegistry = new MetricRegistry();
+		this.metrics = new MaxwellMetrics(config, this.metricRegistry);
 
 		this.replicationConnectionPool = new C3P0ConnectionPool(
 			config.replicationMysql.getConnectionURI(false),
@@ -95,8 +102,6 @@ public MaxwellContext(MaxwellConfig config) throws SQLException, URISyntaxExcept
 			config.maxwellMysql.user,
 			config.maxwellMysql.password
 		);
-		// do NOT probe the regular maxwell connection pool; the database might not be created yet.
-
 		if ( this.config.initPosition != null )
 			this.initialPosition = this.config.initPosition;
 
@@ -109,6 +114,8 @@ public MaxwellContext(MaxwellConfig config) throws SQLException, URISyntaxExcept
 		this.heartbeatNotifier = new HeartbeatNotifier();
 		List<MaxwellDiagnostic> diagnostics = new ArrayList<>(Collections.singletonList(new BinlogConnectorDiagnostic(this)));
 		this.diagnosticContext = new MaxwellDiagnosticContext(config.diagnosticConfig, diagnostics);
+
+		this.healthCheckRegistry = new HealthCheckRegistry();
 	}
 
 	public MaxwellConfig getConfig() {
@@ -177,9 +184,17 @@ private void sendFinalHeartbeat() {
 	public void shutdown(AtomicBoolean complete) {
 		try {
 			taskManager.stop(this.error);
+			this.metrics.stop();
+
 			this.replicationConnectionPool.release();
+			this.replicationConnectionPool = null;
+
 			this.maxwellConnectionPool.release();
+			this.maxwellConnectionPool = null;
+
 			this.rawMaxwellConnectionPool.release();
+			this.rawMaxwellConnectionPool = null;
+
 			complete.set(true);
 		} catch (Exception e) {
 			LOGGER.error("Exception occurred during shutdown:", e);