Vitess VStream Replicator Support

pom.xml

@@ -173,6 +173,12 @@
       <version>15.3</version>
     </dependency>
 
+    <dependency>
+      <groupId>io.vitess</groupId>
+      <artifactId>vitess-grpc-client</artifactId>
+      <version>14.0.1</version>
+    </dependency>
+
     <!-- utils and support libs -->
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
@@ -261,8 +267,9 @@
       <artifactId>simpleclient_servlet</artifactId>
       <version>0.9.0</version>
     </dependency>
+
     <!-- OpenCensus used for pushing metrics to stackdriver -->
-    <dependency>
+    <!-- <dependency>
         <groupId>io.opencensus</groupId>
         <artifactId>opencensus-contrib-dropwizard</artifactId>
         <version>${opencensus.version}</version>
@@ -282,7 +289,7 @@
         <groupId>io.opencensus</groupId>
         <artifactId>opencensus-exporter-stats-stackdriver</artifactId>
         <version>${opencensus.version}</version>
-    </dependency>
+    </dependency> -->
 
     <!-- producer libs -->
     <dependency>

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

@@ -9,6 +9,7 @@
 import com.zendesk.maxwell.replication.BinlogConnectorReplicator;
 import com.zendesk.maxwell.replication.Position;
 import com.zendesk.maxwell.replication.Replicator;
+import com.zendesk.maxwell.replication.VStreamReplicator;
 import com.zendesk.maxwell.row.HeartbeatRowMap;
 import com.zendesk.maxwell.schema.*;
 import com.zendesk.maxwell.schema.columndef.ColumnDefCastException;
@@ -174,7 +175,7 @@ protected Position getInitialPosition() throws Exception {
 			}
 
 			/* fourth method: capture the current master position. */
-			if ( initial == null ) {
+			if ( initial == null && !config.vitessEnabled ) {
 				try ( Connection c = context.getReplicationConnection() ) {
 					initial = Position.capture(c, config.gtidMode);
 				}
@@ -203,7 +204,8 @@ public String getMaxwellVersion() {
 	static String bootString = "Maxwell v%s is booting (%s), starting at %s";
 	private void logBanner(AbstractProducer producer, Position initialPosition) {
 		String producerName = producer.getClass().getSimpleName();
-		LOGGER.info(String.format(bootString, getMaxwellVersion(), producerName, initialPosition.toString()));
+		String position = initialPosition != null ? initialPosition.toString() : "the latest position";
+		LOGGER.info(String.format(bootString, getMaxwellVersion(), producerName, position));
 	}
 
 	/**
@@ -239,13 +241,8 @@ public void start() throws Exception {
 	}
 
 	private void startInner() throws Exception {
-		try ( Connection connection = this.context.getReplicationConnection();
-		      Connection rawConnection = this.context.getRawMaxwellConnection() ) {
-			MaxwellMysqlStatus.ensureReplicationMysqlState(connection);
-			MaxwellMysqlStatus.ensureMaxwellMysqlState(rawConnection);
-			if (config.gtidMode) {
-				MaxwellMysqlStatus.ensureGtidMysqlState(connection);
-			}
+		try (Connection rawConnection = this.context.getRawMaxwellConnection()) {
+			MaxwellMysqlStatus.ensureMaxwellMysqlState(rawConnection, config.vitessEnabled);
 
 			SchemaStoreSchema.ensureMaxwellSchema(rawConnection, this.config.databaseName);
 
@@ -256,41 +253,65 @@ private void startInner() throws Exception {
 
 		AbstractProducer producer = this.context.getProducer();
 
+		if (!config.vitessEnabled) {
+			try (Connection connection = context.getReplicationConnection()) {
+				MaxwellMysqlStatus.ensureReplicationMysqlState(connection);
+
+				if (config.gtidMode) {
+					MaxwellMysqlStatus.ensureGtidMysqlState(connection);
+				}
+			}
+		}
+
 		Position initPosition = getInitialPosition();
 		logBanner(producer, initPosition);
 		this.context.setPosition(initPosition);
 
-		MysqlSchemaStore mysqlSchemaStore = new MysqlSchemaStore(this.context, initPosition);
-		BootstrapController bootstrapController = this.context.getBootstrapController(mysqlSchemaStore.getSchemaID());
+		if (config.vitessEnabled) {
+			this.replicator = new VStreamReplicator(
+				config.vitessConfig,
+				producer,
+				context.getPositionStore(),
+				initPosition,
+				context.getMetrics(),
+				context.getFilter(),
+				config.bufferMemoryUsage,
+				config.binlogEventQueueSize
+			);
+		} else {
+			MysqlSchemaStore mysqlSchemaStore = new MysqlSchemaStore(this.context, initPosition);
+			BootstrapController bootstrapController = this.context
+					.getBootstrapController(mysqlSchemaStore.getSchemaID());
+
+			this.context.startSchemaCompactor();
 
-		this.context.startSchemaCompactor();
+			if (config.recaptureSchema) {
+				mysqlSchemaStore.captureAndSaveSchema();
+			}
 
-		if (config.recaptureSchema) {
-			mysqlSchemaStore.captureAndSaveSchema();
-		}
+			mysqlSchemaStore.getSchema(); // trigger schema to load / capture before we start the replicator.
 
-		mysqlSchemaStore.getSchema(); // trigger schema to load / capture before we start the replicator.
-
-		this.replicator = new BinlogConnectorReplicator(
-			mysqlSchemaStore,
-			producer,
-			bootstrapController,
-			config.replicationMysql,
-			config.replicaServerID,
-			config.databaseName,
-			context.getMetrics(),
-			initPosition,
-			false,
-			config.clientID,
-			context.getHeartbeatNotifier(),
-			config.scripting,
-			context.getFilter(),
-			context.getConfig().getIgnoreMissingSchema(),
-			config.outputConfig,
-			config.bufferMemoryUsage,
-			config.replicationReconnectionRetries,
-			config.binlogEventQueueSize
-		);
+			this.replicator = new BinlogConnectorReplicator(
+				mysqlSchemaStore,
+				producer,
+				bootstrapController,
+				config.replicationMysql,
+				config.replicaServerID,
+				config.databaseName,
+				context.getMetrics(),
+				initPosition,
+				false,
+				config.clientID,
+				context.getHeartbeatNotifier(),
+				config.scripting,
+				context.getFilter(),
+				context.getConfig().getIgnoreMissingSchema(),
+				config.outputConfig,
+				config.bufferMemoryUsage,
+				config.replicationReconnectionRetries,
+				config.binlogEventQueueSize
+			);
+		}
 
 		context.setReplicator(replicator);
 		this.context.start();

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

@@ -81,6 +81,16 @@ public class MaxwellConfig extends AbstractConfig {
 	 */
 	public Boolean gtidMode;
 
+	/**
+	 * If Maxwell is running against a vitess cluster.
+	 */
+	public Boolean vitessEnabled;
+
+	/**
+	 * Vitess (vtgate) connection config
+	 */
+	public MaxwellVitessConfig vitessConfig;
+
 	/**
 	 * Name of database in which to store maxwell data (default `maxwell`)
 	 */
@@ -641,6 +651,10 @@ public MaxwellConfig() { // argv is only null in tests
 		this.schemaMysql = new MaxwellMysqlConfig();
 		this.masterRecovery = false;
 		this.gtidMode = false;
+
+		this.vitessEnabled = false;
+		this.vitessConfig = new MaxwellVitessConfig();
+
 		this.bufferedProducerSize = 200;
 		this.outputConfig = new MaxwellOutputConfig();
 		setup(null, null); // setup defaults
@@ -1043,10 +1057,12 @@ private void setup(OptionSet options, Properties properties) {
 		this.replicationMysql   = parseMysqlConfig("replication_", options, properties);
 		this.schemaMysql        = parseMysqlConfig("schema_", options, properties);
 		this.gtidMode           = fetchBooleanOption("gtid_mode", options, properties, System.getenv(GTID_MODE_ENV) != null);
-
 		this.databaseName       = fetchStringOption("schema_database", options, properties, "maxwell");
 		this.maxwellMysql.database = this.databaseName;
 
+		this.vitessEnabled = fetchBooleanOption("vitess", options, properties, false);
+		this.vitessConfig = parseVitessConfig(options, properties);
+
 		this.producerFactory    = fetchProducerFactory(options, properties);
 		this.producerType       = fetchStringOption("producer", options, properties, "stdout");
 		this.producerAckTimeout = fetchLongOption("producer_ack_timeout", options, properties, 0L);

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

@@ -14,6 +14,7 @@
 import com.zendesk.maxwell.schema.MysqlSchemaCompactor;
 import com.zendesk.maxwell.schema.PositionStoreThread;
 import com.zendesk.maxwell.schema.ReadOnlyMysqlPositionStore;
+import com.zendesk.maxwell.schema.VitessPositionStore;
 import com.zendesk.maxwell.util.C3P0ConnectionPool;
 import com.zendesk.maxwell.util.RunLoopProcess;
 import com.zendesk.maxwell.util.StoppableTask;
@@ -127,10 +128,13 @@ public MaxwellContext(MaxwellConfig config) throws SQLException, URISyntaxExcept
 		if ( this.config.initPosition != null )
 			this.initialPosition = this.config.initPosition;
 
-		if ( this.config.replayMode ) {
-			this.positionStore = new ReadOnlyMysqlPositionStore(this.getMaxwellConnectionPool(), this.getServerID(), this.config.clientID, config.gtidMode);
+		ConnectionPool cp = getMaxwellConnectionPool();
+		if (this.config.replayMode) {
+			this.positionStore = new ReadOnlyMysqlPositionStore(cp, getServerID(), config.clientID, config.gtidMode);
+		} else if (config.vitessEnabled) {
+			this.positionStore = new VitessPositionStore(cp, config.clientID);
 		} else {
-			this.positionStore = new MysqlPositionStore(this.getMaxwellConnectionPool(), this.getServerID(), this.config.clientID, config.gtidMode);
+			this.positionStore = new MysqlPositionStore(cp, getServerID(), config.clientID, config.gtidMode);
 		}
 
 		this.heartbeatNotifier = new HeartbeatNotifier();
@@ -255,7 +259,7 @@ public Thread terminate(Exception error) {
 		}
 
 		if (taskManager.requestStop()) {
-			if (this.error == null && this.replicator != null) {
+			if (this.error == null && this.replicator != null && !config.vitessEnabled) {
 				sendFinalHeartbeat();
 			}
 			this.terminationThread = spawnTerminateThread();

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

@@ -109,13 +109,18 @@ public static void ensureReplicationMysqlState(Connection c) throws SQLException
 	/**
 	 * Verify that the maxwell database is in the expected state
 	 * @param c a JDBC connection
+	 * @param vitessEnabled whether or not we are running in a vitess environment
 	 * @throws SQLException if we have database issues
 	 * @throws MaxwellCompatibilityError if we're not in the expected state
 	 */
-	public static void ensureMaxwellMysqlState(Connection c) throws SQLException, MaxwellCompatibilityError {
+	public static void ensureMaxwellMysqlState(Connection c, boolean vitessEnabled) throws SQLException, MaxwellCompatibilityError {
 		MaxwellMysqlStatus m = new MaxwellMysqlStatus(c);
 
-		m.ensureVariableState("read_only", "OFF");
+		// Vitess reports read_only=ON while running in a vttestserver and sometimes even
+		// in production. We need to ignore this setting when running against Vitess.
+		if (!vitessEnabled) {
+			m.ensureVariableState("read_only", "OFF");
+		}
 	}
 
 	/**

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

@@ -0,0 +1,35 @@
+package com.zendesk.maxwell;
+
+public class MaxwellVitessConfig {
+	public String vtgateHost;
+	public int vtgatePort;
+
+	public String user;
+	public String password;
+
+	public String keyspace;
+	public String shard;
+
+	public boolean usePlaintext;
+	public String tlsCA;
+	public String tlsCert;
+	public String tlsKey;
+	public String tlsServerName;
+
+	public MaxwellVitessConfig() {
+		this.vtgateHost = "localhost";
+		this.vtgatePort = 15991;
+		this.usePlaintext = true;
+
+		this.tlsCA = null;
+		this.tlsCert = null;
+		this.tlsKey = null;
+		this.tlsServerName = null;
+
+		this.user = null;
+		this.password = null;
+
+		this.keyspace = null;
+		this.shard = "";
+	}
+}

src/main/java/com/zendesk/maxwell/monitoring/MaxwellMetrics.java

@@ -12,7 +12,7 @@
 import com.zendesk.maxwell.MaxwellContext;
 import io.prometheus.client.CollectorRegistry;
 import io.prometheus.client.dropwizard.DropwizardExports;
-import io.opencensus.exporter.stats.stackdriver.StackdriverStatsExporter;
+// import io.opencensus.exporter.stats.stackdriver.StackdriverStatsExporter;
 
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -126,17 +126,18 @@ private void setup(MaxwellConfig config) {
 		}
 
 		if (config.metricsReportingType.contains(reportingTypeStackdriver)) {
-			io.opencensus.metrics.Metrics.getExportComponent().getMetricProducerManager().add(
-				new io.opencensus.contrib.dropwizard.DropWizardMetrics(
-				  Collections.singletonList(this.registry)));
-
-			try {
-				StackdriverStatsExporter.createAndRegister();
-			} catch (java.io.IOException e) {
-				LOGGER.error("Maxwell encountered an error in creating the stackdriver exporter.", e);
-			}
-
-			LOGGER.info("Stackdriver metrics reporter enabled");
+			throw new RuntimeException("Stackdriver metrics reporting is not supported for now");
+			// io.opencensus.metrics.Metrics.getExportComponent().getMetricProducerManager().add(
+			// 	new io.opencensus.contrib.dropwizard.DropWizardMetrics(
+			// 	  Collections.singletonList(this.registry)));
+
+			// try {
+			// 	StackdriverStatsExporter.createAndRegister();
+			// } catch (java.io.IOException e) {
+			// 	LOGGER.error("Maxwell encountered an error in creating the stackdriver exporter.", e);
+			// }
+
+			// LOGGER.info("Stackdriver metrics reporter enabled");
 		}
 
 		if (config.metricsReportingType.contains(reportingTypeHttp)) {

src/main/java/com/zendesk/maxwell/replication/BinlogConnectorReplicator.java

@@ -338,7 +338,8 @@ protected void processRow(RowMap row) throws Exception {
 		if ( row instanceof HeartbeatRowMap) {
 			producer.push(row);
 			if (stopAtHeartbeat != null) {
-				long thisHeartbeat = row.getPosition().getLastHeartbeatRead();
+				Position p = (Position) row.getPosition();
+				long thisHeartbeat = p.getLastHeartbeatRead();
 				if (thisHeartbeat >= stopAtHeartbeat) {
 					LOGGER.info("received final heartbeat " + thisHeartbeat + "; stopping replicator");
 					// terminate runLoop

src/main/java/com/zendesk/maxwell/replication/Position.java

@@ -54,13 +54,11 @@ public String toCommandline() {
 
 	@Override
 	public boolean equals(Object o) {
-		if ( !(o instanceof Position) ) {
-			return false;
+		if (o instanceof Position) {
+			Position other = (Position) o;
+			return lastHeartbeatRead == other.lastHeartbeatRead && binlogPosition.equals(other.binlogPosition);
 		}
-		Position other = (Position) o;
-
-		return lastHeartbeatRead == other.lastHeartbeatRead
-			&& binlogPosition.equals(other.binlogPosition);
+		return false;
 	}
 
 	@Override
@@ -69,8 +67,10 @@ public int hashCode() {
 	}
 
 	public boolean newerThan(Position other) {
-		if ( other == null )
-			return true;
-		return this.getBinlogPosition().newerThan(other.getBinlogPosition());
+		if (other instanceof Position) {
+			Position mOther = (Position) other;
+			return binlogPosition.newerThan(mOther.binlogPosition);
+		}
+		return true;
 	}
 }