Vitess VStream Replicator Support

pom.xml

@@ -178,6 +178,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>
@@ -291,7 +297,6 @@
       <artifactId>jedis</artifactId>
       <version>3.5.1</version>
     </dependency>
-
     <!-- metrics -->
     <dependency>
       <groupId>io.dropwizard.metrics</groupId>

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/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/VStreamObserver.java

@@ -0,0 +1,75 @@
+package com.zendesk.maxwell.replication;
+
+import java.util.List;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import binlogdata.Binlogdata.VEvent;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
+import io.vitess.proto.Vtgate;
+
+///
+// Observes the VStream response stream, extracts events from VStream responses,
+// then passes them to the VStreamReplicator via a queue for processing.
+//
+public class VStreamObserver implements StreamObserver<Vtgate.VStreamResponse> {
+	private static final Logger LOGGER = LoggerFactory.getLogger(VStreamObserver.class);
+	private final AtomicBoolean mustStop = new AtomicBoolean(false);
+	private final LinkedBlockingDeque<VEvent> queue;
+	private Exception lastException = null;
+
+	public VStreamObserver(LinkedBlockingDeque<VEvent> queue) {
+		this.queue = queue;
+	}
+
+	// Shuts down the observer
+	public void stop() {
+		mustStop.set(true);
+	}
+
+	@Override
+	public void onNext(Vtgate.VStreamResponse response) {
+		LOGGER.debug("Received {} VEvents in the VStreamResponse:", response.getEventsCount());
+
+		List<VEvent> messageEvents = response.getEventsList();
+		for (VEvent event : messageEvents) {
+			LOGGER.trace("VEvent: {}", event);
+			enqueueEvent(event);
+		}
+	}
+
+	@Override
+	public void onError(Throwable t) {
+		this.lastException = Status.fromThrowable(t).asException();
+		LOGGER.error("VStream streaming onError. Status: {}", lastException);
+		stop();
+	}
+
+	@Override
+	public void onCompleted() {
+		LOGGER.info("VStream streaming completed.");
+		stop();
+	}
+
+	public Exception getLastException() {
+		return lastException;
+	}
+
+	// Pushes an event to the queue for VStreamReplicator to process.
+	private void enqueueEvent(VEvent event) {
+		while (mustStop.get() != true) {
+			try {
+				if (queue.offer(event, 100, TimeUnit.MILLISECONDS)) {
+					break;
+				}
+			} catch (InterruptedException e) {
+				return;
+			}
+		}
+	}
+}