sub-transaction positioning

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

@@ -1271,7 +1271,7 @@ private void setupInitPosition(OptionSet options) {
 				}
 			}
 
-			this.initPosition = new Position(new BinlogPosition(pos, initPositionSplit[0]), lastHeartbeat);
+			this.initPosition = new Position(new BinlogPosition(pos, initPositionSplit[0]), lastHeartbeat, 0L);
 		}
 	}
 

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

@@ -256,20 +256,26 @@ public Thread terminate(Exception error) {
 
 		if (taskManager.requestStop()) {
 			if (this.error == null && this.replicator != null) {
-				sendFinalHeartbeat();
+				stopReplicator();
 			}
 			this.terminationThread = spawnTerminateThread();
 		}
 		return this.terminationThread;
 	}
 
 
-	private void sendFinalHeartbeat() {
+	/**
+	 * Attempt to gracefully terminate the binlog replicator by sending a heartbeat
+	 * through the system for the replicator to stop at.  If this fails we'll terminate
+	 * more forcefully.
+	 */
+	private void stopReplicator() {
 		long heartbeat = System.currentTimeMillis();
 		LOGGER.info("Sending final heartbeat: " + heartbeat);
 		try {
 			this.replicator.stopAtHeartbeat(heartbeat);
 			this.positionStore.heartbeat(heartbeat);
+
 			long deadline = heartbeat + 5000L;
 			while (positionStoreThread.getPosition().getLastHeartbeatRead() < heartbeat) {
 				if (System.currentTimeMillis() > deadline) {
@@ -423,8 +429,7 @@ public RecoveryInfo getRecoveryInfo() throws SQLException {
 	 * @param r A processed Rowmap
 	 */
 	public void setPosition(RowMap r) {
-		if ( r.isTXCommit() )
-			this.setPosition(r.getNextPosition());
+		this.setPosition(r.getNextPosition());
 	}
 
 	/**

src/main/java/com/zendesk/maxwell/bootstrap/BootstrapController.java

@@ -165,4 +165,8 @@ private void pushSkippedRows() throws Exception {
 		}
 	}
 
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.BINLOG;
+	}
 }

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

@@ -148,4 +148,9 @@ public void requestStop() throws Exception {
 	@Override
 	public void awaitStop(Long timeout) throws TimeoutException {
 	}
+
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.SUPPORT;
+	}
 }

src/main/java/com/zendesk/maxwell/producer/AbstractAsyncProducer.java

@@ -1,7 +1,6 @@
 package com.zendesk.maxwell.producer;
 
 import com.codahale.metrics.Gauge;
-import com.zendesk.maxwell.MaxwellConfig;
 import com.zendesk.maxwell.MaxwellContext;
 import com.zendesk.maxwell.monitoring.Metrics;
 import com.zendesk.maxwell.replication.Position;
@@ -16,21 +15,21 @@ public class CallbackCompleter {
 		private final MaxwellContext context;
 		private final int metricsAgeSloMs;
 		private final Position position;
-		private final boolean isTXCommit;
+		private final boolean shouldCommit;
 		private final long messageID;
 
-		public CallbackCompleter(InflightMessageList inflightMessages, Position position, boolean isTXCommit, MaxwellContext context, long messageID) {
+		public CallbackCompleter(InflightMessageList inflightMessages, Position position, boolean shouldCommit, MaxwellContext context, long messageID) {
 			this.inflightMessages = inflightMessages;
 			this.context = context;
 			this.metricsAgeSloMs = context.getConfig().metricsAgeSlo * 1000;
 			this.position = position;
-			this.isTXCommit = isTXCommit;
+			this.shouldCommit = shouldCommit;
 			this.messageID = messageID;
 		}
 
 		public void markCompleted() {
 			inflightMessages.freeSlot(messageID);
-			if(isTXCommit) {
+			if(shouldCommit) {
 				InflightMessageList.InflightMessage message = inflightMessages.completeMessage(position);
 
 				if (message != null) {

src/main/java/com/zendesk/maxwell/producer/MaxwellBigQueryProducer.java

@@ -239,7 +239,12 @@ public void awaitStop(Long timeout) throws TimeoutException {
     taskState.awaitStop(thread, timeout);
   }
 
-  @Override
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.PRODUCER;
+	}
+
+	@Override
   public void run() {
     this.thread = Thread.currentThread();
     while (true) {
@@ -279,4 +284,4 @@ future, new BigQueryCallback(this, appendContext, cc, r.getNextPosition(),
             this.context),
         MoreExecutors.directExecutor());
   }
-}
+}

src/main/java/com/zendesk/maxwell/producer/MaxwellKafkaProducer.java

@@ -304,6 +304,11 @@ public void awaitStop(Long timeout) throws TimeoutException {
 		taskState.awaitStop(thread, timeout);
 	}
 
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.PRODUCER;
+	}
+
 	// force-close for tests.
 	public void close() {
 		kafka.close();

src/main/java/com/zendesk/maxwell/producer/MaxwellPubsubProducer.java

@@ -275,7 +275,12 @@ public void awaitStop(Long timeout) throws TimeoutException {
     taskState.awaitStop(thread, timeout);
   }
 
-  @Override
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.PRODUCER;
+	}
+
+	@Override
   public StoppableTask getStoppableTask() {
     return this;
   }

src/main/java/com/zendesk/maxwell/producer/MaxwellRedisProducer.java

@@ -2,6 +2,7 @@
 
 import com.zendesk.maxwell.MaxwellContext;
 import com.zendesk.maxwell.row.RowMap;
+import com.zendesk.maxwell.util.TaskManager;
 import com.zendesk.maxwell.util.TopicInterpolator;
 import com.zendesk.maxwell.util.StoppableTask;
 import org.slf4j.Logger;
@@ -20,7 +21,8 @@
 import java.util.HashSet;
 import java.util.Set;
 
-public class MaxwellRedisProducer extends AbstractProducer implements StoppableTask {
+public
+class MaxwellRedisProducer extends AbstractProducer implements StoppableTask {
 	private static final Logger logger = LoggerFactory.getLogger(MaxwellRedisProducer.class);
 	private final String channel;
 	private final String redisType;
@@ -190,6 +192,11 @@ public void requestStop() {
 	@Override
 	public void awaitStop(Long timeout) { }
 
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.PRODUCER;
+	}
+
 	@Override
 	public StoppableTask getStoppableTask() {
 		return this;

src/main/java/com/zendesk/maxwell/producer/StdoutProducer.java

@@ -21,6 +21,7 @@ public void push(RowMap r) throws Exception {
 			System.currentTimeMillis() - r.getTimestampMillis(),
 			TimeUnit.MILLISECONDS
 		);
+
 		this.context.setPosition(r);
 	}
 }

src/main/java/com/zendesk/maxwell/recovery/Recovery.java

@@ -56,7 +56,7 @@ public HeartbeatRowMap recover() throws Exception {
 		List<BinlogPosition> list = getBinlogInfo();
 		for ( int i = list.size() - 1; i >= 0 ; i-- ) {
 			BinlogPosition binlogPosition = list.get(i);
-			Position position = Position.valueOf(binlogPosition, recoveryInfo.getHeartbeat());
+			Position position = Position.valueOf(binlogPosition, recoveryInfo.getHeartbeat(), 0L);
 			Metrics metrics = new NoOpMetrics();
 
 			LOGGER.debug("scanning binlog: {}", binlogPosition);

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

@@ -167,11 +167,24 @@ private RowMap buildRowMap(String type, Position position, Position nextPosition
 		return map;
 	}
 
-	public List<RowMap> jsonMaps(Table table, long lastHeartbeatRead, String rowQuery) throws ColumnDefCastException {
+	public List<RowMap> jsonMaps(Table table, long lastHeartbeatRead, String rowQuery, BinlogPosition lastTableMapPosition) throws ColumnDefCastException {
 		ArrayList<RowMap> list = new ArrayList<>();
 
-		Position position     = Position.valueOf(this.position, lastHeartbeatRead);
-		Position nextPosition = Position.valueOf(this.nextPosition, lastHeartbeatRead);
+		Position position     = Position.valueOf(this.position, lastHeartbeatRead, 0L);
+		/*
+			at the time our RowMaps are created we have no idea where we are
+			in a transaciton; we may be the last row-event in the transaction
+			in which case we're going to "commit" -- ie advance maxwell.positions,
+			or we may be in the middle of a transaction.
+
+			If we're in the middle of a transaction, we want to leave the binlog pointer
+			pointing at the table-map event that precedes this.  The calling function
+			will additionally add a row-offset to the position of each row.
+
+			All this mechanism allows maxwell to stop and restart in the middle of very large
+			transactions.
+		 */
+		Position nextPosition = Position.valueOf(lastTableMapPosition, lastHeartbeatRead, 0L);
 
 		switch ( getType() ) {
 			case WRITE_ROWS:

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

@@ -63,6 +63,7 @@ public class BinlogConnectorReplicator extends RunLoopProcess implements Replica
 
 	private Position lastHeartbeatPosition;
 	private final HeartbeatNotifier heartbeatNotifier;
+	private Position startPosition;
 	private Long stopAtHeartbeat;
 	private Filter filter;
 	private Boolean ignoreMissingSchema;
@@ -86,6 +87,11 @@ public class BinlogConnectorReplicator extends RunLoopProcess implements Replica
 
 	private boolean isConnected = false;
 
+	@Override
+	public StopPriority getStopPriority() {
+		return StopPriority.BINLOG;
+	}
+
 	private class ClientReconnectedException extends Exception {}
 
 	public BinlogConnectorReplicator(
@@ -182,6 +188,7 @@ public BinlogConnectorReplicator(
 		this.client.setUseSendAnnotateRowsEvent(true);
 
 
+		this.startPosition = start;
 		BinlogPosition startBinlog = start.getBinlogPosition();
 		if (startBinlog.getGtidSetStr() != null) {
 			String gtidStr = startBinlog.getGtidSetStr();
@@ -408,8 +415,8 @@ private void processQueryEvent(BinlogConnectorEvent event) throws Exception {
 			data.getDatabase(),
 			data.getSql(),
 			this.schemaStore,
-			Position.valueOf(event.getPosition(), getLastHeartbeatRead()),
-			Position.valueOf(event.getNextPosition(), getLastHeartbeatRead()),
+			Position.valueOf(event.getPosition(), getLastHeartbeatRead(), 0L),
+			Position.valueOf(event.getNextPosition(), getLastHeartbeatRead(), 0L),
 			event.getEvent().getHeader().getTimestamp()
 		);
 	}
@@ -534,11 +541,13 @@ private void tryReconnect() throws TimeoutException {
 	 * @return A RowMapBuffer of rows; either in-memory or on disk.
 	 */
 
-	private RowMapBuffer getTransactionRows(BinlogConnectorEvent beginEvent) throws Exception {
+	private RowMapBuffer getTransactionRows(BinlogConnectorEvent beginEvent, long startTXOffset) throws Exception {
 		BinlogConnectorEvent event;
+		BinlogPosition lastTableMapPosition = null;
 		RowMapBuffer buffer = new RowMapBuffer(MAX_TX_ELEMENTS, this.bufferMemoryUsage);
 
 		String currentQuery = null;
+		long txOffset = 0;
 
 		while ( true ) {
 			event = pollEvent();
@@ -551,7 +560,9 @@ private RowMapBuffer getTransactionRows(BinlogConnectorEvent beginEvent) throws
 			EventType eventType = event.getEvent().getHeader().getEventType();
 			if (event.isCommitEvent()) {
 				if (!buffer.isEmpty()) {
-					buffer.getLast().setTXCommit();
+					RowMap lastRow = buffer.getLast();
+					lastRow.setTXCommit();
+					lastRow.updateNextPosition(event.getNextPosition());
 					long timeSpent = buffer.getLast().getTimestampMillis() - beginEvent.getEvent().getHeader().getTimestamp();
 					transactionExecutionTime.update(timeSpent);
 					transactionRowCount.update(buffer.size());
@@ -574,22 +585,27 @@ private RowMapBuffer getTransactionRows(BinlogConnectorEvent beginEvent) throws
 					if ( table != null && shouldOutputEvent(table.getDatabase(), table.getName(), filter, table.getColumnNames()) ) {
 						List<RowMap> rows;
 						try {
-							rows = event.jsonMaps(table, getLastHeartbeatRead(), currentQuery);
+							rows = event.jsonMaps(table, getLastHeartbeatRead(), currentQuery, lastTableMapPosition);
 						} catch ( ColumnDefCastException e ) {
 							logColumnDefCastException(table, e);
 
 							throw(e);
 						}
 
-						for ( RowMap r : rows )
-							if (shouldOutputRowMap(table.getDatabase(), table.getName(), r, filter)) {
-								buffer.add(r);
+						for ( RowMap r : rows ) {
+							r.setXoffset(txOffset++);
+							if ( shouldOutputRowMap(table.getDatabase(), table.getName(), r, filter) ) {
+								if ( startTXOffset <= r.getXoffset() ) {
+									buffer.add(r);
+								}
 							}
+						}
 					}
 					break;
 				case TABLE_MAP:
 					TableMapEventData data = event.tableMapData();
 					tableCache.processEvent(getSchema(), this.filter, this.ignoreMissingSchema, data.getTableId(), data.getDatabase(), data.getTable());
+					lastTableMapPosition = event.getPosition();
 					break;
 				case ROWS_QUERY:
 					RowsQueryEventData rqed = event.getEvent().getData();
@@ -663,6 +679,7 @@ private void logColumnDefCastException(Table table, ColumnDefCastException e) {
 	 * @return either a RowMap or null
 	 */
 	public RowMap getRow() throws Exception {
+		long txOffset;
 		BinlogConnectorEvent event;
 
 		if ( stopOnEOF && hitEOF )
@@ -699,29 +716,46 @@ public RowMap getRow() throws Exception {
 				}
 			}
 
+
 			switch (event.getType()) {
 				case WRITE_ROWS:
 				case EXT_WRITE_ROWS:
 				case UPDATE_ROWS:
 				case EXT_UPDATE_ROWS:
 				case DELETE_ROWS:
 				case EXT_DELETE_ROWS:
-					LOGGER.warn("Started replication stream inside a transaction.  This shouldn't normally happen.");
-					LOGGER.warn("Assuming new transaction at unexpected event:" + event);
+					LOGGER.warn("Started replication stream inside a transaction, probably recovering from stopping inside transaction.");
 
 					queue.offerFirst(event);
-					rowBuffer = getTransactionRows(event);
+					rowBuffer = getTransactionRows(event, 0);
 					break;
 				case TABLE_MAP:
 					TableMapEventData data = event.tableMapData();
 					tableCache.processEvent(getSchema(), this.filter,this.ignoreMissingSchema, data.getTableId(), data.getDatabase(), data.getTable());
+
+					if ( startPosition != null && startPosition.getTXOffset() > 0 ) {
+						LOGGER.info("Restarting maxwell inside transaction: {}", startPosition);
+						queue.offerFirst(event); // put table map back on top of queue
+						rowBuffer = getTransactionRows(event, startPosition.getTXOffset());
+					}
+
+					startPosition = null;
 					break;
 				case QUERY:
+					txOffset = 0;
 					QueryEventData qe = event.queryData();
 					String sql = qe.getSql();
+
+					if ( startPosition != null && startPosition.getTXOffset() > 0 ) {
+						LOGGER.info("Restarting maxwell inside transaction: {}", startPosition);
+						txOffset = startPosition.getTXOffset();;
+					}
+
+					startPosition = null;
+
 					if (BinlogConnectorEvent.BEGIN.equals(sql)) {
 						try {
-							rowBuffer = getTransactionRows(event);
+							rowBuffer = getTransactionRows(event, txOffset);
 						} catch ( ClientReconnectedException e ) {
 							// rowBuffer should already be empty by the time we get to this switch
 							// statement, but we null it for clarity
@@ -739,8 +773,17 @@ public RowMap getRow() throws Exception {
 					// in mariaDB the GTID event supplants the normal BEGIN
 					MariadbGtidEventData g = event.mariaGtidData();
 					if ( (g.getFlags() & MariadbGtidEventData.FL_STANDALONE) == 0 ) {
+						txOffset = 0;
+
+						if ( startPosition != null && startPosition.getTXOffset() > 0 ) {
+							LOGGER.info("Restarting maxwell inside transaction: {}", startPosition);
+							txOffset = startPosition.getTXOffset();;
+						}
+
+						startPosition = null;
+
 						try {
-							rowBuffer = getTransactionRows(event);
+							rowBuffer = getTransactionRows(event, txOffset);
 						} catch ( ClientReconnectedException e ) {
 							// rowBuffer should already be empty by the time we get to this switch
 							// statement, but we null it for clarity