Merge pull request #292 from zendesk/heartbeats

consume OpenReplicator heartbeats

pom.xml

@@ -65,7 +65,7 @@
     <dependency>
     	<groupId>com.zendesk</groupId>
     	<artifactId>open-replicator</artifactId>
-    	<version>1.4.0</version>
+    	<version>1.4.2</version>
     </dependency>
     <dependency>
     	<groupId>net.sf.jopt-simple</groupId>

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

@@ -58,6 +58,8 @@ public MaxwellReplicator(Schema currentSchema, AbstractProducer producer, Abstra
 
 		this.replicator.setLevel2BufferSize(50 * 1024 * 1024);
 
+		this.replicator.setHeartbeatPeriod(0.5f);
+
 		this.producer = producer;
 		this.bootstrapper = bootstrapper;
 
@@ -79,6 +81,13 @@ private void ensureReplicatorThread() throws Exception {
 			LOGGER.warn("open-replicator stopped at position " + replicator.getBinlogFileName() + ":" + replicator.getBinlogPosition() + " -- restarting");
 			replicator.start();
 		}
+
+		Long ms = replicator.millisSinceLastEvent();
+		if ( ms != null && ms > 2000 ) {
+			LOGGER.warn("no heartbeat heard from server in " + ms + "ms.  restarting replication.");
+			replicator.stop(5, TimeUnit.SECONDS);
+			replicator.start();
+		}
 	}
 
 	@Override