diff --git a/go/base/context.go b/go/base/context.go
index f90171698..934e67e96 100644
--- a/go/base/context.go
+++ b/go/base/context.go
@@ -166,6 +166,9 @@ type MigrationContext struct {
 	CutOverType                  CutOver
 	ReplicaServerId              uint
 
+	// Number of workers used by the trx coordinator
+	NumWorkers int
+
 	Hostname                               string
 	AssumeMasterHostname                   string
 	ApplierTimeZone                        string
diff --git a/go/binlog/gomysql_reader.go b/go/binlog/gomysql_reader.go
index d42ba1f30..30c72c74b 100644
--- a/go/binlog/gomysql_reader.go
+++ b/go/binlog/gomysql_reader.go
@@ -6,12 +6,10 @@
 package binlog
 
 import (
-	"fmt"
 	"sync"
 
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/mysql"
-	"github.com/github/gh-ost/go/sql"
 
 	"time"
 
@@ -76,68 +74,17 @@ func (this *GoMySQLReader) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinate
 	return &returnCoordinates
 }
 
-// StreamEvents
-func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error {
-	if this.currentCoordinates.IsLogPosOverflowBeyond4Bytes(&this.LastAppliedRowsEventHint) {
-		return fmt.Errorf("Unexpected rows event at %+v, the binlog end_log_pos is overflow 4 bytes", this.currentCoordinates)
-	}
-
-	if this.currentCoordinates.SmallerThanOrEquals(&this.LastAppliedRowsEventHint) {
-		this.migrationContext.Log.Debugf("Skipping handled query at %+v", this.currentCoordinates)
-		return nil
-	}
-
-	dml := ToEventDML(ev.Header.EventType.String())
-	if dml == NotDML {
-		return fmt.Errorf("Unknown DML type: %s", ev.Header.EventType.String())
-	}
-	for i, row := range rowsEvent.Rows {
-		if dml == UpdateDML && i%2 == 1 {
-			// An update has two rows (WHERE+SET)
-			// We do both at the same time
-			continue
-		}
-		binlogEntry := NewBinlogEntryAt(this.currentCoordinates)
-		binlogEntry.DmlEvent = NewBinlogDMLEvent(
-			string(rowsEvent.Table.Schema),
-			string(rowsEvent.Table.Table),
-			dml,
-		)
-		switch dml {
-		case InsertDML:
-			{
-				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(row)
-			}
-		case UpdateDML:
-			{
-				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
-				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(rowsEvent.Rows[i+1])
-			}
-		case DeleteDML:
-			{
-				binlogEntry.DmlEvent.WhereColumnValues = sql.ToColumnValues(row)
-			}
-		}
-		// The channel will do the throttling. Whoever is reading from the channel
-		// decides whether action is taken synchronously (meaning we wait before
-		// next iteration) or asynchronously (we keep pushing more events)
-		// In reality, reads will be synchronous
-		entriesChannel <- binlogEntry
-	}
-	this.LastAppliedRowsEventHint = this.currentCoordinates
-	return nil
-}
-
-// StreamEvents
-func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChannel chan<- *BinlogEntry) error {
-	if canStopStreaming() {
-		return nil
-	}
+// StreamEvents reads binlog events and sends them to the given channel.
+// It is blocking and should be executed in a goroutine.
+func (this *GoMySQLReader) StreamEvents(ctx context.Context, canStopStreaming func() bool, eventChannel chan<- *replication.BinlogEvent) error {
 	for {
 		if canStopStreaming() {
-			break
+			return nil
+		}
+		if err := ctx.Err(); err != nil {
+			return err
 		}
-		ev, err := this.binlogStreamer.GetEvent(context.Background())
+		ev, err := this.binlogStreamer.GetEvent(ctx)
 		if err != nil {
 			return err
 		}
@@ -147,24 +94,8 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 			this.currentCoordinates.LogPos = int64(ev.Header.LogPos)
 			this.currentCoordinates.EventSize = int64(ev.Header.EventSize)
 		}()
-
-		switch binlogEvent := ev.Event.(type) {
-		case *replication.RotateEvent:
-			func() {
-				this.currentCoordinatesMutex.Lock()
-				defer this.currentCoordinatesMutex.Unlock()
-				this.currentCoordinates.LogFile = string(binlogEvent.NextLogName)
-			}()
-			this.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", this.currentCoordinates.LogFile, int64(ev.Header.LogPos), binlogEvent.NextLogName)
-		case *replication.RowsEvent:
-			if err := this.handleRowsEvent(ev, binlogEvent, entriesChannel); err != nil {
-				return err
-			}
-		}
+		eventChannel <- ev
 	}
-	this.migrationContext.Log.Debugf("done streaming events")
-
-	return nil
 }
 
 func (this *GoMySQLReader) Close() error {
diff --git a/go/cmd/gh-ost/main.go b/go/cmd/gh-ost/main.go
index 0829429e0..fcb0181fe 100644
--- a/go/cmd/gh-ost/main.go
+++ b/go/cmd/gh-ost/main.go
@@ -109,6 +109,7 @@ func main() {
 	defaultRetries := flag.Int64("default-retries", 60, "Default number of retries for various operations before panicking")
 	cutOverLockTimeoutSeconds := flag.Int64("cut-over-lock-timeout-seconds", 3, "Max number of seconds to hold locks on tables while attempting to cut-over (retry attempted when lock exceeds timeout) or attempting instant DDL")
 	niceRatio := flag.Float64("nice-ratio", 0, "force being 'nice', imply sleep time per chunk time; range: [0.0..100.0]. Example values: 0 is aggressive. 1: for every 1ms spent copying rows, sleep additional 1ms (effectively doubling runtime); 0.7: for every 10ms spend in a rowcopy chunk, spend 7ms sleeping immediately after")
+	flag.IntVar(&migrationContext.NumWorkers, "workers", 8, "Number of concurrent workers for applying DML events. Each worker uses one goroutine.")
 
 	maxLagMillis := flag.Int64("max-lag-millis", 1500, "replication lag at which to throttle operation")
 	replicationLagQuery := flag.String("replication-lag-query", "", "Deprecated. gh-ost uses an internal, subsecond resolution query")
diff --git a/go/logic/applier.go b/go/logic/applier.go
index 59562dc7f..f5aa65cd2 100644
--- a/go/logic/applier.go
+++ b/go/logic/applier.go
@@ -78,12 +78,14 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
 	}
 }
 
-func (this *Applier) InitDBConnections() (err error) {
+func (this *Applier) InitDBConnections(maxConns int) (err error) {
 	applierUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
 	uriWithMulti := fmt.Sprintf("%s&multiStatements=true", applierUri)
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, uriWithMulti); err != nil {
 		return err
 	}
+	this.db.SetMaxOpenConns(maxConns)
+	this.db.SetMaxIdleConns(maxConns)
 	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
 	if this.singletonDB, _, err = mysql.GetDB(this.migrationContext.Uuid, singletonApplierUri); err != nil {
 		return err
diff --git a/go/logic/applier_test.go b/go/logic/applier_test.go
index f53e65ffb..e2f138e8b 100644
--- a/go/logic/applier_test.go
+++ b/go/logic/applier_test.go
@@ -270,7 +270,7 @@ func (suite *ApplierTestSuite) TestInitDBConnections() {
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	suite.Require().Equal("8.0.40", migrationContext.ApplierMySQLVersion)
@@ -313,7 +313,7 @@ func (suite *ApplierTestSuite) TestApplyDMLEventQueries() {
 	suite.Require().NoError(applier.prepareQueries())
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	dmlEvents := []*binlog.BinlogDMLEvent{
@@ -373,7 +373,7 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTables() {
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -408,7 +408,7 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTablesWithGhostTableExi
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -442,7 +442,7 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTablesWithGhostTableExi
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.ValidateOrDropExistingTables()
@@ -483,7 +483,7 @@ func (suite *ApplierTestSuite) TestCreateGhostTable() {
 	applier := NewApplier(migrationContext)
 	defer applier.Teardown()
 
-	err = applier.InitDBConnections()
+	err = applier.InitDBConnections(8)
 	suite.Require().NoError(err)
 
 	err = applier.CreateGhostTable()
diff --git a/go/logic/coordinator.go b/go/logic/coordinator.go
new file mode 100644
index 000000000..1921daffc
--- /dev/null
+++ b/go/logic/coordinator.go
@@ -0,0 +1,555 @@
+package logic
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"errors"
+
+	"github.com/github/gh-ost/go/base"
+	"github.com/github/gh-ost/go/binlog"
+	"github.com/github/gh-ost/go/mysql"
+	"github.com/github/gh-ost/go/sql"
+	"github.com/go-mysql-org/go-mysql/replication"
+)
+
+type Coordinator struct {
+	migrationContext *base.MigrationContext
+
+	binlogReader *binlog.GoMySQLReader
+
+	onChangelogEvent func(dmlEvent *binlog.BinlogDMLEvent) error
+
+	applier *Applier
+
+	throttler *Throttler
+
+	// Atomic counter for number of active workers (not in workerQueue)
+	busyWorkers atomic.Int64
+
+	// Mutex protecting currentCoordinates
+	currentCoordinatesMutex sync.Mutex
+	// The binlog coordinates of the low water mark transaction.
+	currentCoordinates mysql.BinlogCoordinates
+
+	// Mutex to protect the fields below
+	mu sync.Mutex
+
+	// list of workers
+	workers []*Worker
+
+	// The low water mark. This is the sequence number of the last job that has been committed.
+	lowWaterMark int64
+
+	// This is a map of completed jobs by their sequence numbers.
+	// This is used when updating the low water mark.
+	// It records the binlog coordinates of the completed transaction.
+	completedJobs map[int64]*mysql.BinlogCoordinates
+
+	// These are the jobs that are waiting for a previous job to complete.
+	// They are indexed by the sequence number of the job they are waiting for.
+	waitingJobs map[int64][]chan struct{}
+
+	events chan *replication.BinlogEvent
+
+	workerQueue chan *Worker
+
+	finishedMigrating atomic.Bool
+}
+
+// Worker takes jobs from the Coordinator and applies the job's DML events.
+type Worker struct {
+	id          int
+	coordinator *Coordinator
+	eventQueue  chan *replication.BinlogEvent
+
+	executedJobs     atomic.Int64
+	dmlEventsApplied atomic.Int64
+	waitTimeNs       atomic.Int64
+	busyTimeNs       atomic.Int64
+}
+
+type stats struct {
+	dmlRate float64
+	trxRate float64
+
+	// Number of DML events applied
+	dmlEventsApplied int64
+
+	// Number of transactions processed
+	executedJobs int64
+
+	// Time spent applying DML events
+	busyTime time.Duration
+
+	// Time spent waiting on transaction dependecies
+	// or waiting on events to arrive in queue.
+	waitTime time.Duration
+}
+
+func (w *Worker) ProcessEvents() error {
+	databaseName := w.coordinator.migrationContext.DatabaseName
+	originalTableName := w.coordinator.migrationContext.OriginalTableName
+	changelogTableName := w.coordinator.migrationContext.GetChangelogTableName()
+
+	for {
+		if w.coordinator.finishedMigrating.Load() {
+			return nil
+		}
+
+		// Wait for first event
+		waitStart := time.Now()
+		ev := <-w.eventQueue
+		w.waitTimeNs.Add(time.Since(waitStart).Nanoseconds())
+
+		// Verify this is a GTID Event
+		gtidEvent, ok := ev.Event.(*replication.GTIDEvent)
+		if !ok {
+			w.coordinator.migrationContext.Log.Debugf("Received unexpected event: %v\n", ev)
+		}
+
+		// Wait for conditions to be met
+		waitChannel := w.coordinator.WaitForTransaction(gtidEvent.LastCommitted)
+		if waitChannel != nil {
+			waitStart := time.Now()
+			<-waitChannel
+			timeWaited := time.Since(waitStart)
+			w.waitTimeNs.Add(timeWaited.Nanoseconds())
+		}
+
+		// Process the transaction
+		var changelogEvent *binlog.BinlogDMLEvent
+		dmlEvents := make([]*binlog.BinlogDMLEvent, 0, int(atomic.LoadInt64(&w.coordinator.migrationContext.DMLBatchSize)))
+	events:
+		for {
+			// wait for next event in the transaction
+			waitStart := time.Now()
+			ev := <-w.eventQueue
+			w.waitTimeNs.Add(time.Since(waitStart).Nanoseconds())
+
+			if ev == nil {
+				fmt.Printf("Worker %d ending transaction early\n", w.id)
+				break events
+			}
+
+			switch binlogEvent := ev.Event.(type) {
+			case *replication.RowsEvent:
+				// Is this an event that we're interested in?
+				// We're only interested in events that:
+				// * affect the table we're migrating
+				// * affect the changelog table
+
+				dml := binlog.ToEventDML(ev.Header.EventType.String())
+				if dml == binlog.NotDML {
+					return fmt.Errorf("unknown DML type: %s", ev.Header.EventType.String())
+				}
+
+				if !strings.EqualFold(databaseName, string(binlogEvent.Table.Schema)) {
+					continue
+				}
+
+				if !strings.EqualFold(originalTableName, string(binlogEvent.Table.Table)) && !strings.EqualFold(changelogTableName, string(binlogEvent.Table.Table)) {
+					continue
+				}
+
+				for i, row := range binlogEvent.Rows {
+					if dml == binlog.UpdateDML && i%2 == 1 {
+						// An update has two rows (WHERE+SET)
+						// We do both at the same time
+						continue
+					}
+					dmlEvent := binlog.NewBinlogDMLEvent(
+						string(binlogEvent.Table.Schema),
+						string(binlogEvent.Table.Table),
+						dml,
+					)
+					switch dml {
+					case binlog.InsertDML:
+						{
+							dmlEvent.NewColumnValues = sql.ToColumnValues(row)
+						}
+					case binlog.UpdateDML:
+						{
+							dmlEvent.WhereColumnValues = sql.ToColumnValues(row)
+							dmlEvent.NewColumnValues = sql.ToColumnValues(binlogEvent.Rows[i+1])
+						}
+					case binlog.DeleteDML:
+						{
+							dmlEvent.WhereColumnValues = sql.ToColumnValues(row)
+						}
+					}
+
+					if strings.EqualFold(changelogTableName, string(binlogEvent.Table.Table)) {
+						// If this is a change on the changelog table, queue it up to be processed after
+						// the end of the transaction.
+						changelogEvent = dmlEvent
+					} else {
+						dmlEvents = append(dmlEvents, dmlEvent)
+
+						if len(dmlEvents) == cap(dmlEvents) {
+							if err := w.applyDMLEvents(dmlEvents); err != nil {
+								w.coordinator.migrationContext.Log.Errore(err)
+							}
+							dmlEvents = dmlEvents[:0]
+						}
+					}
+				}
+			case *replication.XIDEvent:
+				if len(dmlEvents) > 0 {
+					if err := w.applyDMLEvents(dmlEvents); err != nil {
+						w.coordinator.migrationContext.Log.Errore(err)
+					}
+				}
+
+				w.executedJobs.Add(1)
+				break events
+			}
+		}
+
+		w.coordinator.MarkTransactionCompleted(gtidEvent.SequenceNumber, int64(ev.Header.LogPos), int64(ev.Header.EventSize))
+
+		// Did we see a changelog event?
+		// Handle it now
+		if changelogEvent != nil {
+			// wait for all transactions before this point
+			waitChannel = w.coordinator.WaitForTransaction(gtidEvent.SequenceNumber - 1)
+			if waitChannel != nil {
+				waitStart := time.Now()
+				<-waitChannel
+				w.waitTimeNs.Add(time.Since(waitStart).Nanoseconds())
+			}
+			w.coordinator.HandleChangeLogEvent(changelogEvent)
+		}
+
+		w.coordinator.workerQueue <- w
+		w.coordinator.busyWorkers.Add(-1)
+	}
+}
+
+func (w *Worker) applyDMLEvents(dmlEvents []*binlog.BinlogDMLEvent) error {
+	if w.coordinator.throttler != nil {
+		w.coordinator.throttler.throttle(nil)
+	}
+	busyStart := time.Now()
+	err := w.coordinator.applier.ApplyDMLEventQueries(dmlEvents)
+	if err != nil {
+		//TODO(meiji163) add retry
+		return err
+	}
+	w.busyTimeNs.Add(time.Since(busyStart).Nanoseconds())
+	w.dmlEventsApplied.Add(int64(len(dmlEvents)))
+	return nil
+}
+
+func NewCoordinator(migrationContext *base.MigrationContext, applier *Applier, throttler *Throttler, onChangelogEvent func(dmlEvent *binlog.BinlogDMLEvent) error) *Coordinator {
+	return &Coordinator{
+		migrationContext: migrationContext,
+
+		onChangelogEvent: onChangelogEvent,
+
+		throttler: throttler,
+
+		currentCoordinates: mysql.BinlogCoordinates{},
+
+		binlogReader: binlog.NewGoMySQLReader(migrationContext),
+
+		lowWaterMark:  0,
+		completedJobs: make(map[int64]*mysql.BinlogCoordinates),
+		waitingJobs:   make(map[int64][]chan struct{}),
+
+		events: make(chan *replication.BinlogEvent, 1000),
+	}
+}
+
+func (c *Coordinator) StartStreaming(ctx context.Context, canStopStreaming func() bool) error {
+	coords := c.GetCurrentBinlogCoordinates()
+	err := c.binlogReader.ConnectBinlogStreamer(*coords)
+	if err != nil {
+		return err
+	}
+	defer c.binlogReader.Close()
+
+	var retries int64
+	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		if canStopStreaming() {
+			return nil
+		}
+		if err := c.binlogReader.StreamEvents(ctx, canStopStreaming, c.events); err != nil {
+			if errors.Is(err, context.Canceled) {
+				return err
+			}
+
+			c.migrationContext.Log.Infof("StreamEvents encountered unexpected error: %+v", err)
+			c.migrationContext.MarkPointOfInterest()
+
+			if retries >= c.migrationContext.MaxRetries() {
+				return fmt.Errorf("%d successive failures in streamer reconnect at coordinates %+v", retries, coords)
+			}
+			c.migrationContext.Log.Infof("Reconnecting... Will resume at %+v", coords)
+
+			// We reconnect at the position of the last low water mark.
+			// Some jobs after the low water mark may have already applied, but
+			// it's OK to reapply them since the DML operations are idempotent.
+			coords := c.GetCurrentBinlogCoordinates()
+			if err := c.binlogReader.ConnectBinlogStreamer(*coords); err != nil {
+				return err
+			}
+			retries += 1
+		}
+	}
+}
+
+func (c *Coordinator) ProcessEventsUntilNextChangelogEvent() (*binlog.BinlogDMLEvent, error) {
+	databaseName := c.migrationContext.DatabaseName
+	changelogTableName := c.migrationContext.GetChangelogTableName()
+
+	for ev := range c.events {
+		switch binlogEvent := ev.Event.(type) {
+		case *replication.RowsEvent:
+			dml := binlog.ToEventDML(ev.Header.EventType.String())
+			if dml == binlog.NotDML {
+				return nil, fmt.Errorf("unknown DML type: %s", ev.Header.EventType.String())
+			}
+
+			if !strings.EqualFold(databaseName, string(binlogEvent.Table.Schema)) {
+				continue
+			}
+
+			if !strings.EqualFold(changelogTableName, string(binlogEvent.Table.Table)) {
+				continue
+			}
+
+			for i, row := range binlogEvent.Rows {
+				if dml == binlog.UpdateDML && i%2 == 1 {
+					// An update has two rows (WHERE+SET)
+					// We do both at the same time
+					continue
+				}
+				dmlEvent := binlog.NewBinlogDMLEvent(
+					string(binlogEvent.Table.Schema),
+					string(binlogEvent.Table.Table),
+					dml,
+				)
+				switch dml {
+				case binlog.InsertDML:
+					{
+						dmlEvent.NewColumnValues = sql.ToColumnValues(row)
+					}
+				case binlog.UpdateDML:
+					{
+						dmlEvent.WhereColumnValues = sql.ToColumnValues(row)
+						dmlEvent.NewColumnValues = sql.ToColumnValues(binlogEvent.Rows[i+1])
+					}
+				case binlog.DeleteDML:
+					{
+						dmlEvent.WhereColumnValues = sql.ToColumnValues(row)
+					}
+				}
+
+				return dmlEvent, nil
+			}
+		}
+	}
+
+	//nolint:nilnil
+	return nil, nil
+}
+
+// ProcessEventsUntilDrained reads binlog events and sends them to the workers to process.
+// It exits when the event queue is empty and all the workers are returned to the workerQueue.
+func (c *Coordinator) ProcessEventsUntilDrained() error {
+	for {
+		select {
+		// Read events from the binlog and submit them to the next worker
+		case ev := <-c.events:
+			{
+				if c.finishedMigrating.Load() {
+					return nil
+				}
+
+				switch binlogEvent := ev.Event.(type) {
+				case *replication.GTIDEvent:
+					if c.lowWaterMark == 0 && binlogEvent.SequenceNumber > 0 {
+						c.lowWaterMark = binlogEvent.SequenceNumber - 1
+					}
+				case *replication.RotateEvent:
+					c.currentCoordinatesMutex.Lock()
+					c.currentCoordinates.LogFile = string(binlogEvent.NextLogName)
+					c.currentCoordinatesMutex.Unlock()
+					c.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", c.currentCoordinates.LogFile, int64(ev.Header.LogPos), binlogEvent.NextLogName)
+					continue
+				default: // ignore all other events
+					continue
+				}
+
+				worker := <-c.workerQueue
+				c.busyWorkers.Add(1)
+
+				worker.eventQueue <- ev
+
+				ev = <-c.events
+
+				switch binlogEvent := ev.Event.(type) {
+				case *replication.QueryEvent:
+					if bytes.Equal([]byte("BEGIN"), binlogEvent.Query) {
+					} else {
+						worker.eventQueue <- nil
+						continue
+					}
+				default:
+					worker.eventQueue <- nil
+					continue
+				}
+
+			events:
+				for {
+					ev = <-c.events
+					switch ev.Event.(type) {
+					case *replication.RowsEvent:
+						worker.eventQueue <- ev
+					case *replication.XIDEvent:
+						worker.eventQueue <- ev
+
+						// We're done with this transaction
+						break events
+					}
+				}
+			}
+
+		// No events in the queue. Check if all workers are sleeping now
+		default:
+			{
+				if c.busyWorkers.Load() == 0 {
+					return nil
+				}
+			}
+		}
+	}
+}
+
+func (c *Coordinator) InitializeWorkers(count int) {
+	c.workerQueue = make(chan *Worker, count)
+	for i := 0; i < count; i++ {
+		w := &Worker{id: i, coordinator: c, eventQueue: make(chan *replication.BinlogEvent, 1000)}
+
+		c.mu.Lock()
+		c.workers = append(c.workers, w)
+		c.mu.Unlock()
+
+		c.workerQueue <- w
+		go w.ProcessEvents()
+	}
+}
+
+// GetWorkerStats collects profiling stats for ProcessEvents from each worker.
+func (c *Coordinator) GetWorkerStats() []stats {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	statSlice := make([]stats, 0, len(c.workers))
+	for _, w := range c.workers {
+		stat := stats{}
+		stat.dmlEventsApplied = w.dmlEventsApplied.Load()
+		stat.executedJobs = w.executedJobs.Load()
+		stat.busyTime = time.Duration(w.busyTimeNs.Load())
+		stat.waitTime = time.Duration(w.waitTimeNs.Load())
+		if stat.busyTime.Milliseconds() > 0 {
+			stat.dmlRate = 1000.0 * float64(stat.dmlEventsApplied) / float64(stat.busyTime.Milliseconds())
+			stat.trxRate = 1000.0 * float64(stat.executedJobs) / float64(stat.busyTime.Milliseconds())
+		}
+		statSlice = append(statSlice, stat)
+	}
+	return statSlice
+}
+
+func (c *Coordinator) WaitForTransaction(lastCommitted int64) chan struct{} {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if lastCommitted <= c.lowWaterMark {
+		return nil
+	}
+
+	if _, ok := c.completedJobs[lastCommitted]; ok {
+		return nil
+	}
+
+	waitChannel := make(chan struct{})
+	c.waitingJobs[lastCommitted] = append(c.waitingJobs[lastCommitted], waitChannel)
+
+	return waitChannel
+}
+
+func (c *Coordinator) HandleChangeLogEvent(event *binlog.BinlogDMLEvent) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.onChangelogEvent(event)
+}
+
+func (c *Coordinator) MarkTransactionCompleted(sequenceNumber, logPos, eventSize int64) {
+	var channelsToNotify []chan struct{}
+	var lastCoords *mysql.BinlogCoordinates
+
+	func() {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+
+		//c.migrationContext.Log.Infof("Coordinator: Marking job as completed: %d\n", sequenceNumber)
+
+		// Mark the job as completed
+		c.completedJobs[sequenceNumber] = &mysql.BinlogCoordinates{LogPos: logPos, EventSize: eventSize}
+
+		// Then, update the low water mark if possible
+		for {
+			if coords, ok := c.completedJobs[c.lowWaterMark+1]; ok {
+				lastCoords = coords
+				c.lowWaterMark++
+				delete(c.completedJobs, c.lowWaterMark)
+			} else {
+				break
+			}
+		}
+		channelsToNotify = make([]chan struct{}, 0)
+
+		// Schedule any jobs that were waiting for this job to complete
+		for waitingForSequenceNumber, channels := range c.waitingJobs {
+			if waitingForSequenceNumber <= c.lowWaterMark {
+				channelsToNotify = append(channelsToNotify, channels...)
+				delete(c.waitingJobs, waitingForSequenceNumber)
+			}
+		}
+	}()
+
+	// update the binlog coords of the low water mark
+	if lastCoords != nil {
+		func() {
+			// c.migrationContext.Log.Infof("Updating binlog coordinates to %s:%d\n", c.currentCoordinates.LogFile, c.currentCoordinates.LogPos)
+			c.currentCoordinatesMutex.Lock()
+			defer c.currentCoordinatesMutex.Unlock()
+			c.currentCoordinates.LogPos = lastCoords.LogPos
+			c.currentCoordinates.EventSize = lastCoords.EventSize
+		}()
+	}
+
+	for _, waitChannel := range channelsToNotify {
+		waitChannel <- struct{}{}
+	}
+}
+
+func (c *Coordinator) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinates {
+	c.currentCoordinatesMutex.Lock()
+	defer c.currentCoordinatesMutex.Unlock()
+	returnCoordinates := c.currentCoordinates
+	return &returnCoordinates
+}
+
+func (c *Coordinator) Teardown() {
+	c.finishedMigrating.Store(true)
+}
diff --git a/go/logic/coordinator_test.go b/go/logic/coordinator_test.go
new file mode 100644
index 000000000..6a5f6d8df
--- /dev/null
+++ b/go/logic/coordinator_test.go
@@ -0,0 +1,206 @@
+package logic
+
+import (
+	"context"
+	gosql "database/sql"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"path/filepath"
+	"runtime"
+
+	"github.com/github/gh-ost/go/base"
+	"github.com/github/gh-ost/go/binlog"
+	"github.com/github/gh-ost/go/mysql"
+	"github.com/github/gh-ost/go/sql"
+	"github.com/stretchr/testify/suite"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+type CoordinatorTestSuite struct {
+	suite.Suite
+
+	mysqlContainer testcontainers.Container
+	db             *gosql.DB
+}
+
+func (suite *CoordinatorTestSuite) SetupSuite() {
+	ctx := context.Background()
+	req := testcontainers.ContainerRequest{
+		Image:        "mysql:8.0.40",
+		Env:          map[string]string{"MYSQL_ROOT_PASSWORD": "root-password"},
+		WaitingFor:   wait.ForListeningPort("3306/tcp"),
+		ExposedPorts: []string{"3306/tcp"},
+	}
+
+	mysqlContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	})
+	suite.Require().NoError(err)
+
+	suite.mysqlContainer = mysqlContainer
+
+	dsn, err := GetDSN(ctx, mysqlContainer)
+	suite.Require().NoError(err)
+
+	db, err := gosql.Open("mysql", dsn)
+	suite.Require().NoError(err)
+
+	suite.db = db
+}
+
+func (suite *CoordinatorTestSuite) SetupTest() {
+	ctx := context.Background()
+	_, err := suite.db.ExecContext(ctx, "RESET MASTER")
+	suite.Require().NoError(err)
+
+	_, err = suite.db.ExecContext(ctx, "SET @@GLOBAL.binlog_transaction_dependency_tracking = WRITESET")
+	suite.Require().NoError(err)
+
+	_, err = suite.db.ExecContext(ctx, "CREATE DATABASE test")
+	suite.Require().NoError(err)
+}
+
+func (suite *CoordinatorTestSuite) TearDownTest() {
+	ctx := context.Background()
+	_, err := suite.db.ExecContext(ctx, "DROP DATABASE test")
+	suite.Require().NoError(err)
+}
+
+func (suite *CoordinatorTestSuite) TeardownSuite() {
+	ctx := context.Background()
+
+	suite.Assert().NoError(suite.db.Close())
+	suite.Assert().NoError(suite.mysqlContainer.Terminate(ctx))
+}
+
+func (suite *CoordinatorTestSuite) TestApplyDML() {
+	ctx := context.Background()
+
+	connectionConfig, err := GetConnectionConfig(ctx, suite.mysqlContainer)
+	suite.Require().NoError(err)
+
+	_ = os.Remove("/tmp/gh-ost.sock")
+
+	_, err = suite.db.Exec("CREATE TABLE test.gh_ost_test (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255)) ENGINE=InnoDB")
+	suite.Require().NoError(err)
+
+	_, err = suite.db.Exec("CREATE TABLE test._gh_ost_test_gho (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(255))")
+	suite.Require().NoError(err)
+
+	migrationContext := base.NewMigrationContext()
+	migrationContext.DatabaseName = "test"
+	migrationContext.OriginalTableName = "gh_ost_test"
+	migrationContext.AlterStatement = "ALTER TABLE gh_ost_test ENGINE=InnoDB"
+	migrationContext.AllowedRunningOnMaster = true
+	migrationContext.ReplicaServerId = 99999
+	migrationContext.HeartbeatIntervalMilliseconds = 100
+	migrationContext.ThrottleHTTPIntervalMillis = 100
+	migrationContext.DMLBatchSize = 10
+
+	migrationContext.ApplierConnectionConfig = connectionConfig
+	migrationContext.InspectorConnectionConfig = connectionConfig
+
+	migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "name"})
+	migrationContext.GhostTableColumns = sql.NewColumnList([]string{"id", "name"})
+	migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "name"})
+	migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "name"})
+	migrationContext.UniqueKey = &sql.UniqueKey{
+		Name:            "PRIMARY",
+		Columns:         *sql.NewColumnList([]string{"id"}),
+		IsAutoIncrement: true,
+	}
+
+	migrationContext.SetConnectionConfig("innodb")
+	migrationContext.SkipPortValidation = true
+	migrationContext.NumWorkers = 4
+
+	//nolint:dogsled
+	_, filename, _, _ := runtime.Caller(0)
+	migrationContext.ServeSocketFile = filepath.Join(filepath.Dir(filename), "../../tmp/gh-ost.sock")
+
+	applier := NewApplier(migrationContext)
+	err = applier.InitDBConnections(migrationContext.NumWorkers)
+	suite.Require().NoError(err)
+
+	err = applier.prepareQueries()
+	suite.Require().NoError(err)
+
+	err = applier.CreateChangelogTable()
+	suite.Require().NoError(err)
+
+	//  TODO: use errgroup
+	for i := 0; i < 100; i++ {
+		tx, err := suite.db.Begin()
+		suite.Require().NoError(err)
+
+		for j := 0; j < 100; j++ {
+			_, err = tx.Exec("INSERT INTO test.gh_ost_test (name) VALUES ('test')")
+			suite.Require().NoError(err)
+		}
+
+		err = tx.Commit()
+		suite.Require().NoError(err)
+	}
+
+	_, err = suite.db.Exec("UPDATE test.gh_ost_test SET name = 'foobar' WHERE id = 1")
+	suite.Require().NoError(err)
+
+	_, err = suite.db.Exec("INSERT INTO test.gh_ost_test (name) VALUES ('test')")
+	suite.Require().NoError(err)
+
+	_, err = applier.WriteChangelogState("completed")
+	suite.Require().NoError(err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	coord := NewCoordinator(migrationContext, applier, nil,
+		func(dmlEvent *binlog.BinlogDMLEvent) error {
+			fmt.Printf("Received Changelog DML event: %+v\n", dmlEvent)
+			fmt.Printf("Rowdata: %v - %v\n", dmlEvent.NewColumnValues, dmlEvent.WhereColumnValues)
+
+			cancel()
+
+			return nil
+		})
+	coord.applier = applier
+	coord.currentCoordinates = mysql.BinlogCoordinates{
+		LogFile: "binlog.000001",
+		LogPos:  int64(4),
+	}
+	coord.InitializeWorkers(4)
+
+	streamCtx, cancelStreaming := context.WithCancel(context.Background())
+	canStopStreaming := func() bool {
+		return streamCtx.Err() != nil
+	}
+	go func() {
+		err = coord.StartStreaming(streamCtx, canStopStreaming)
+		suite.Require().Equal(context.Canceled, err)
+	}()
+
+	// Give streamer some time to start
+	time.Sleep(1 * time.Second)
+
+	startAt := time.Now()
+
+	for {
+		if ctx.Err() != nil {
+			cancelStreaming()
+			break
+		}
+
+		err = coord.ProcessEventsUntilDrained()
+		suite.Require().NoError(err)
+	}
+
+	fmt.Printf("Time taken: %s\n", time.Since(startAt))
+}
+
+func TestCoordinator(t *testing.T) {
+	suite.Run(t, new(CoordinatorTestSuite))
+}
diff --git a/go/logic/inspect.go b/go/logic/inspect.go
index ea8c3adca..3220cc1ab 100644
--- a/go/logic/inspect.go
+++ b/go/logic/inspect.go
@@ -860,6 +860,28 @@ func (this *Inspector) readChangelogState(hint string) (string, error) {
 	return result, err
 }
 
+// readCurrentBinlogCoordinates reads master status from hooked server
+func (this *Inspector) readCurrentBinlogCoordinates() (*mysql.BinlogCoordinates, error) {
+	var coords *mysql.BinlogCoordinates
+	query := `show /* gh-ost readCurrentBinlogCoordinates */ master status`
+	foundMasterStatus := false
+	err := sqlutils.QueryRowsMap(this.db, query, func(m sqlutils.RowMap) error {
+		coords = &mysql.BinlogCoordinates{
+			LogFile: m.GetString("File"),
+			LogPos:  m.GetInt64("Position"),
+		}
+		foundMasterStatus = true
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	if !foundMasterStatus {
+		return nil, fmt.Errorf("Got no results from SHOW MASTER STATUS. Bailing out")
+	}
+	return coords, nil
+}
+
 func (this *Inspector) getMasterConnectionConfig() (applierConfig *mysql.ConnectionConfig, err error) {
 	this.migrationContext.Log.Infof("Recursively searching for replication master")
 	visitedKeys := mysql.NewInstanceKeyMap()
diff --git a/go/logic/migrator.go b/go/logic/migrator.go
index 751b54a08..24ec3a4d8 100644
--- a/go/logic/migrator.go
+++ b/go/logic/migrator.go
@@ -11,11 +11,12 @@ import (
 	"fmt"
 	"io"
 	"math"
-	"os"
 	"strings"
 	"sync/atomic"
 	"time"
 
+	"os"
+
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/binlog"
 	"github.com/github/gh-ost/go/mysql"
@@ -41,21 +42,6 @@ func ReadChangelogState(s string) ChangelogState {
 
 type tableWriteFunc func() error
 
-type applyEventStruct struct {
-	writeFunc *tableWriteFunc
-	dmlEvent  *binlog.BinlogDMLEvent
-}
-
-func newApplyEventStructByFunc(writeFunc *tableWriteFunc) *applyEventStruct {
-	result := &applyEventStruct{writeFunc: writeFunc}
-	return result
-}
-
-func newApplyEventStructByDML(dmlEvent *binlog.BinlogDMLEvent) *applyEventStruct {
-	result := &applyEventStruct{dmlEvent: dmlEvent}
-	return result
-}
-
 type PrintStatusRule int
 
 const (
@@ -72,7 +58,6 @@ type Migrator struct {
 	parser           *sql.AlterTableParser
 	inspector        *Inspector
 	applier          *Applier
-	eventsStreamer   *EventsStreamer
 	server           *Server
 	throttler        *Throttler
 	hooksExecutor    *HooksExecutor
@@ -86,12 +71,12 @@ type Migrator struct {
 	rowCopyCompleteFlag int64
 	// copyRowsQueue should not be buffered; if buffered some non-damaging but
 	//  excessive work happens at the end of the iteration as new copy-jobs arrive before realizing the copy is complete
-	copyRowsQueue    chan tableWriteFunc
-	applyEventsQueue chan *applyEventStruct
+	copyRowsQueue chan tableWriteFunc
 
 	handledChangelogStates map[string]bool
 
 	finishedMigrating int64
+	trxCoordinator    *Coordinator
 }
 
 func NewMigrator(context *base.MigrationContext, appVersion string) *Migrator {
@@ -100,13 +85,12 @@ func NewMigrator(context *base.MigrationContext, appVersion string) *Migrator {
 		hooksExecutor:              NewHooksExecutor(context),
 		migrationContext:           context,
 		parser:                     sql.NewAlterTableParser(),
-		ghostTableMigrated:         make(chan bool),
+		ghostTableMigrated:         make(chan bool, 1),
 		firstThrottlingCollected:   make(chan bool, 3),
 		rowCopyComplete:            make(chan error),
 		allEventsUpToLockProcessed: make(chan string),
 
 		copyRowsQueue:          make(chan tableWriteFunc),
-		applyEventsQueue:       make(chan *applyEventStruct, base.MaxEventsBatchSize),
 		handledChangelogStates: make(map[string]bool),
 		finishedMigrating:      0,
 	}
@@ -221,17 +205,13 @@ func (this *Migrator) onChangelogStateEvent(dmlEvent *binlog.BinlogDMLEvent) (er
 	case GhostTableMigrated:
 		this.ghostTableMigrated <- true
 	case AllEventsUpToLockProcessed:
-		var applyEventFunc tableWriteFunc = func() error {
-			this.allEventsUpToLockProcessed <- changelogStateString
-			return nil
-		}
 		// at this point we know all events up to lock have been read from the streamer,
 		// because the streamer works sequentially. So those events are either already handled,
 		// or have event functions in applyEventsQueue.
 		// So as not to create a potential deadlock, we write this func to applyEventsQueue
 		// asynchronously, understanding it doesn't really matter.
 		go func() {
-			this.applyEventsQueue <- newApplyEventStructByFunc(&applyEventFunc)
+			this.allEventsUpToLockProcessed <- changelogStateString
 		}()
 	default:
 		return fmt.Errorf("Unknown changelog state: %+v", changelogState)
@@ -350,12 +330,18 @@ func (this *Migrator) Migrate() (err error) {
 	if err := this.initiateInspector(); err != nil {
 		return err
 	}
+
+	this.trxCoordinator = NewCoordinator(this.migrationContext, this.applier, this.throttler, this.onChangelogEvent)
+
 	if err := this.initiateStreaming(); err != nil {
 		return err
 	}
 	if err := this.initiateApplier(); err != nil {
 		return err
 	}
+
+	this.trxCoordinator.applier = this.applier
+
 	if err := this.createFlagFiles(); err != nil {
 		return err
 	}
@@ -378,9 +364,27 @@ func (this *Migrator) Migrate() (err error) {
 		}
 	}
 
+	this.migrationContext.Log.Infof("starting %d applier workers", this.migrationContext.NumWorkers)
+	this.trxCoordinator.InitializeWorkers(this.migrationContext.NumWorkers)
+
 	initialLag, _ := this.inspector.getReplicationLag()
 	this.migrationContext.Log.Infof("Waiting for ghost table to be migrated. Current lag is %+v", initialLag)
-	<-this.ghostTableMigrated
+
+waitForGhostTable:
+	for {
+		select {
+		case <-this.ghostTableMigrated:
+			break waitForGhostTable
+		default:
+			dmlEvent, err := this.trxCoordinator.ProcessEventsUntilNextChangelogEvent()
+			if err != nil {
+				return err
+			}
+
+			this.onChangelogEvent(dmlEvent)
+		}
+	}
+
 	this.migrationContext.Log.Debugf("ghost table migrated")
 	// Yay! We now know the Ghost and Changelog tables are good to examine!
 	// When running on replica, this means the replica has those tables. When running
@@ -406,9 +410,7 @@ func (this *Migrator) Migrate() (err error) {
 	if err := this.countTableRows(); err != nil {
 		return err
 	}
-	if err := this.addDMLEventsListener(); err != nil {
-		return err
-	}
+
 	if err := this.applier.ReadMigrationRangeValues(); err != nil {
 		return err
 	}
@@ -418,6 +420,7 @@ func (this *Migrator) Migrate() (err error) {
 	if err := this.hooksExecutor.onBeforeRowCopy(); err != nil {
 		return err
 	}
+
 	go this.executeWriteFuncs()
 	go this.iterateChunks()
 	this.migrationContext.MarkRowCopyStartTime()
@@ -430,6 +433,7 @@ func (this *Migrator) Migrate() (err error) {
 		return err
 	}
 	this.printStatus(ForcePrintStatusRule)
+	this.printWorkerStats()
 
 	if this.migrationContext.IsCountingTableRows() {
 		this.migrationContext.Log.Info("stopping query for exact row count, because that can accidentally lock out the cut over")
@@ -736,10 +740,13 @@ func (this *Migrator) atomicCutOver() (err error) {
 
 // initiateServer begins listening on unix socket/tcp for incoming interactive commands
 func (this *Migrator) initiateServer() (err error) {
-	var f printStatusFunc = func(rule PrintStatusRule, writer io.Writer) {
+	var printStatus printStatusFunc = func(rule PrintStatusRule, writer io.Writer) {
 		this.printStatus(rule, writer)
 	}
-	this.server = NewServer(this.migrationContext, this.hooksExecutor, f)
+	var printWorkers printWorkersFunc = func(writer io.Writer) {
+		this.printWorkerStats(writer)
+	}
+	this.server = NewServer(this.migrationContext, this.hooksExecutor, printStatus, printWorkers)
 	if err := this.server.BindSocketFile(); err != nil {
 		return err
 	}
@@ -1014,6 +1021,29 @@ func (this *Migrator) shouldPrintMigrationStatusHint(rule PrintStatusRule, elaps
 	return shouldPrint
 }
 
+// printWorkerStats prints cumulative stats from the trxCoordinator workers.
+func (this *Migrator) printWorkerStats(writers ...io.Writer) {
+	writers = append(writers, os.Stdout)
+	mw := io.MultiWriter(writers...)
+
+	busyWorkers := this.trxCoordinator.busyWorkers.Load()
+	totalWorkers := cap(this.trxCoordinator.workerQueue)
+	fmt.Fprintf(mw, "# %d/%d workers are busy\n", busyWorkers, totalWorkers)
+
+	stats := this.trxCoordinator.GetWorkerStats()
+	for id, stat := range stats {
+		fmt.Fprintf(mw,
+			"Worker %d; Waited: %s; Busy: %s; DML Applied: %d (%.2f/s), Trx Applied: %d (%.2f/s)\n",
+			id,
+			base.PrettifyDurationOutput(stat.waitTime),
+			base.PrettifyDurationOutput(stat.busyTime),
+			stat.dmlEventsApplied,
+			stat.dmlRate,
+			stat.executedJobs,
+			stat.trxRate)
+	}
+}
+
 // printStatus prints the progress status, and optionally additionally detailed
 // dump of configuration.
 // `rule` indicates the type of output expected.
@@ -1052,12 +1082,12 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 		return
 	}
 
-	currentBinlogCoordinates := *this.eventsStreamer.GetCurrentBinlogCoordinates()
+	currentBinlogCoordinates := *this.trxCoordinator.GetCurrentBinlogCoordinates()
 
 	status := fmt.Sprintf("Copy: %d/%d %.1f%%; Applied: %d; Backlog: %d/%d; Time: %+v(total), %+v(copy); streamer: %+v; Lag: %.2fs, HeartbeatLag: %.2fs, State: %s; ETA: %s",
 		totalRowsCopied, rowsEstimate, progressPct,
 		atomic.LoadInt64(&this.migrationContext.TotalDMLEventsApplied),
-		len(this.applyEventsQueue), cap(this.applyEventsQueue),
+		len(this.trxCoordinator.events), cap(this.trxCoordinator.events),
 		base.PrettifyDurationOutput(elapsedTime), base.PrettifyDurationOutput(this.migrationContext.ElapsedRowCopyTime()),
 		currentBinlogCoordinates,
 		this.migrationContext.GetCurrentLagDuration().Seconds(),
@@ -1088,22 +1118,16 @@ func (this *Migrator) printStatus(rule PrintStatusRule, writers ...io.Writer) {
 
 // initiateStreaming begins streaming of binary log events and registers listeners for such events
 func (this *Migrator) initiateStreaming() error {
-	this.eventsStreamer = NewEventsStreamer(this.migrationContext)
-	if err := this.eventsStreamer.InitDBConnections(); err != nil {
+	initialCoords, err := this.inspector.readCurrentBinlogCoordinates()
+	if err != nil {
 		return err
 	}
-	this.eventsStreamer.AddListener(
-		false,
-		this.migrationContext.DatabaseName,
-		this.migrationContext.GetChangelogTableName(),
-		func(dmlEvent *binlog.BinlogDMLEvent) error {
-			return this.onChangelogEvent(dmlEvent)
-		},
-	)
+	this.trxCoordinator.currentCoordinates = *initialCoords
 
 	go func() {
-		this.migrationContext.Log.Debugf("Beginning streaming")
-		err := this.eventsStreamer.StreamEvents(this.canStopStreaming)
+		this.migrationContext.Log.Debugf("Beginning streaming at coordinates: %+v", *initialCoords)
+		ctx := context.TODO()
+		err := this.trxCoordinator.StartStreaming(ctx, this.canStopStreaming)
 		if err != nil {
 			this.migrationContext.PanicAbort <- err
 		}
@@ -1117,27 +1141,11 @@ func (this *Migrator) initiateStreaming() error {
 			if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 				return
 			}
-			this.migrationContext.SetRecentBinlogCoordinates(*this.eventsStreamer.GetCurrentBinlogCoordinates())
 		}
 	}()
 	return nil
 }
 
-// addDMLEventsListener begins listening for binlog events on the original table,
-// and creates & enqueues a write task per such event.
-func (this *Migrator) addDMLEventsListener() error {
-	err := this.eventsStreamer.AddListener(
-		false,
-		this.migrationContext.DatabaseName,
-		this.migrationContext.OriginalTableName,
-		func(dmlEvent *binlog.BinlogDMLEvent) error {
-			this.applyEventsQueue <- newApplyEventStructByDML(dmlEvent)
-			return nil
-		},
-	)
-	return err
-}
-
 // initiateThrottler kicks in the throttling collection and the throttling checks.
 func (this *Migrator) initiateThrottler() {
 	this.throttler = NewThrottler(this.migrationContext, this.applier, this.inspector, this.appVersion)
@@ -1153,7 +1161,7 @@ func (this *Migrator) initiateThrottler() {
 
 func (this *Migrator) initiateApplier() error {
 	this.applier = NewApplier(this.migrationContext)
-	if err := this.applier.InitDBConnections(); err != nil {
+	if err := this.applier.InitDBConnections(this.migrationContext.NumWorkers); err != nil {
 		return err
 	}
 	if err := this.applier.ValidateOrDropExistingTables(); err != nil {
@@ -1261,57 +1269,6 @@ func (this *Migrator) iterateChunks() error {
 	}
 }
 
-func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
-	handleNonDMLEventStruct := func(eventStruct *applyEventStruct) error {
-		if eventStruct.writeFunc != nil {
-			if err := this.retryOperation(*eventStruct.writeFunc); err != nil {
-				return this.migrationContext.Log.Errore(err)
-			}
-		}
-		return nil
-	}
-	if eventStruct.dmlEvent == nil {
-		return handleNonDMLEventStruct(eventStruct)
-	}
-	if eventStruct.dmlEvent != nil {
-		dmlEvents := [](*binlog.BinlogDMLEvent){}
-		dmlEvents = append(dmlEvents, eventStruct.dmlEvent)
-		var nonDmlStructToApply *applyEventStruct
-
-		availableEvents := len(this.applyEventsQueue)
-		batchSize := int(atomic.LoadInt64(&this.migrationContext.DMLBatchSize))
-		if availableEvents > batchSize-1 {
-			// The "- 1" is because we already consumed one event: the original event that led to this function getting called.
-			// So, if DMLBatchSize==1 we wish to not process any further events
-			availableEvents = batchSize - 1
-		}
-		for i := 0; i < availableEvents; i++ {
-			additionalStruct := <-this.applyEventsQueue
-			if additionalStruct.dmlEvent == nil {
-				// Not a DML. We don't group this, and we don't batch any further
-				nonDmlStructToApply = additionalStruct
-				break
-			}
-			dmlEvents = append(dmlEvents, additionalStruct.dmlEvent)
-		}
-		// Create a task to apply the DML event; this will be execute by executeWriteFuncs()
-		var applyEventFunc tableWriteFunc = func() error {
-			return this.applier.ApplyDMLEventQueries(dmlEvents)
-		}
-		if err := this.retryOperation(applyEventFunc); err != nil {
-			return this.migrationContext.Log.Errore(err)
-		}
-		if nonDmlStructToApply != nil {
-			// We pulled DML events from the queue, and then we hit a non-DML event. Wait!
-			// We need to handle it!
-			if err := handleNonDMLEventStruct(nonDmlStructToApply); err != nil {
-				return this.migrationContext.Log.Errore(err)
-			}
-		}
-	}
-	return nil
-}
-
 // executeWriteFuncs writes data via applier: both the rowcopy and the events backlog.
 // This is where the ghost table gets the data. The function fills the data single-threaded.
 // Both event backlog and rowcopy events are polled; the backlog events have precedence.
@@ -1320,6 +1277,7 @@ func (this *Migrator) executeWriteFuncs() error {
 		this.migrationContext.Log.Debugf("Noop operation; not really executing write funcs")
 		return nil
 	}
+
 	for {
 		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 			return nil
@@ -1327,40 +1285,34 @@ func (this *Migrator) executeWriteFuncs() error {
 
 		this.throttler.throttle(nil)
 
-		// We give higher priority to event processing, then secondary priority to
-		// rowcopy
+		// We give higher priority to event processing.
+		// ProcessEventsUntilDrained will process all events in the queue, and then return once no more events are available.
+		this.trxCoordinator.ProcessEventsUntilDrained()
+
+		this.throttler.throttle(nil)
+
+		// And secondary priority to rowcopy
 		select {
-		case eventStruct := <-this.applyEventsQueue:
+		case copyRowsFunc := <-this.copyRowsQueue:
 			{
-				if err := this.onApplyEventStruct(eventStruct); err != nil {
-					return err
+				copyRowsStartTime := time.Now()
+				// Retries are handled within the copyRowsFunc
+				if err := copyRowsFunc(); err != nil {
+					return this.migrationContext.Log.Errore(err)
+				}
+				if niceRatio := this.migrationContext.GetNiceRatio(); niceRatio > 0 {
+					copyRowsDuration := time.Since(copyRowsStartTime)
+					sleepTimeNanosecondFloat64 := niceRatio * float64(copyRowsDuration.Nanoseconds())
+					sleepTime := time.Duration(int64(sleepTimeNanosecondFloat64)) * time.Nanosecond
+					time.Sleep(sleepTime)
 				}
 			}
 		default:
 			{
-				select {
-				case copyRowsFunc := <-this.copyRowsQueue:
-					{
-						copyRowsStartTime := time.Now()
-						// Retries are handled within the copyRowsFunc
-						if err := copyRowsFunc(); err != nil {
-							return this.migrationContext.Log.Errore(err)
-						}
-						if niceRatio := this.migrationContext.GetNiceRatio(); niceRatio > 0 {
-							copyRowsDuration := time.Since(copyRowsStartTime)
-							sleepTimeNanosecondFloat64 := niceRatio * float64(copyRowsDuration.Nanoseconds())
-							sleepTime := time.Duration(int64(sleepTimeNanosecondFloat64)) * time.Nanosecond
-							time.Sleep(sleepTime)
-						}
-					}
-				default:
-					{
-						// Hmmmmm... nothing in the queue; no events, but also no row copy.
-						// This is possible upon load. Let's just sleep it over.
-						this.migrationContext.Log.Debugf("Getting nothing in the write queue. Sleeping...")
-						time.Sleep(time.Second)
-					}
-				}
+				// Hmmmmm... nothing in the queue; no events, but also no row copy.
+				// This is possible upon load. Let's just sleep it over.
+				this.migrationContext.Log.Debugf("Getting nothing in the write queue. Sleeping...")
+				time.Sleep(time.Second)
 			}
 		}
 	}
@@ -1383,10 +1335,6 @@ func (this *Migrator) finalCleanup() error {
 			this.migrationContext.Log.Errore(err)
 		}
 	}
-	if err := this.eventsStreamer.Close(); err != nil {
-		this.migrationContext.Log.Errore(err)
-	}
-
 	if err := this.retryOperation(this.applier.DropChangelogTable); err != nil {
 		return err
 	}
@@ -1412,6 +1360,16 @@ func (this *Migrator) finalCleanup() error {
 func (this *Migrator) teardown() {
 	atomic.StoreInt64(&this.finishedMigrating, 1)
 
+	if this.trxCoordinator != nil {
+		this.migrationContext.Log.Infof("Tearing down coordinator")
+		this.trxCoordinator.Teardown()
+	}
+
+	if this.throttler != nil {
+		this.migrationContext.Log.Infof("Tearing down throttler")
+		this.throttler.Teardown()
+	}
+
 	if this.inspector != nil {
 		this.migrationContext.Log.Infof("Tearing down inspector")
 		this.inspector.Teardown()
@@ -1421,14 +1379,4 @@ func (this *Migrator) teardown() {
 		this.migrationContext.Log.Infof("Tearing down applier")
 		this.applier.Teardown()
 	}
-
-	if this.eventsStreamer != nil {
-		this.migrationContext.Log.Infof("Tearing down streamer")
-		this.eventsStreamer.Teardown()
-	}
-
-	if this.throttler != nil {
-		this.migrationContext.Log.Infof("Tearing down throttler")
-		this.throttler.Teardown()
-	}
 }
diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go
index dfdfe5390..35b001cae 100644
--- a/go/logic/migrator_test.go
+++ b/go/logic/migrator_test.go
@@ -7,17 +7,17 @@ package logic
 
 import (
 	"context"
-	gosql "database/sql"
 	"errors"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	gosql "database/sql"
+
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	"github.com/testcontainers/testcontainers-go"
@@ -46,29 +46,29 @@ func TestMigratorOnChangelogEvent(t *testing.T) {
 		}))
 	})
 
-	t.Run("state-AllEventsUpToLockProcessed", func(t *testing.T) {
-		var wg sync.WaitGroup
-		wg.Add(1)
-		go func(wg *sync.WaitGroup) {
-			defer wg.Done()
-			es := <-migrator.applyEventsQueue
-			require.NotNil(t, es)
-			require.NotNil(t, es.writeFunc)
-		}(&wg)
-
-		columnValues := sql.ToColumnValues([]interface{}{
-			123,
-			time.Now().Unix(),
-			"state",
-			AllEventsUpToLockProcessed,
-		})
-		require.Nil(t, migrator.onChangelogEvent(&binlog.BinlogDMLEvent{
-			DatabaseName:    "test",
-			DML:             binlog.InsertDML,
-			NewColumnValues: columnValues,
-		}))
-		wg.Wait()
-	})
+	// t.Run("state-AllEventsUpToLockProcessed", func(t *testing.T) {
+	// 	var wg sync.WaitGroup
+	// 	wg.Add(1)
+	// 	go func(wg *sync.WaitGroup) {
+	// 		defer wg.Done()
+	// 		es := <-migrator.applyEventsQueue
+	// 		require.NotNil(t, es)
+	// 		require.NotNil(t, es.writeFunc)
+	// 	}(&wg)
+
+	// 	columnValues := sql.ToColumnValues([]interface{}{
+	// 		123,
+	// 		time.Now().Unix(),
+	// 		"state",
+	// 		AllEventsUpToLockProcessed,
+	// 	})
+	// 	require.Nil(t, migrator.onChangelogEvent(&binlog.BinlogDMLEvent{
+	// 		DatabaseName:    "test",
+	// 		DML:             binlog.InsertDML,
+	// 		NewColumnValues: columnValues,
+	// 	}))
+	// 	wg.Wait()
+	// })
 
 	t.Run("state-GhostTableMigrated", func(t *testing.T) {
 		go func() {
@@ -290,6 +290,7 @@ func (suite *MigratorTestSuite) SetupSuite() {
 		ContainerRequest: req,
 		Started:          true,
 	})
+
 	suite.Require().NoError(err)
 
 	suite.mysqlContainer = mysqlContainer
@@ -313,7 +314,10 @@ func (suite *MigratorTestSuite) TeardownSuite() {
 func (suite *MigratorTestSuite) SetupTest() {
 	ctx := context.Background()
 
-	_, err := suite.db.ExecContext(ctx, "CREATE DATABASE test")
+	_, err := suite.db.ExecContext(ctx, "SET @@GLOBAL.binlog_transaction_dependency_tracking = WRITESET")
+	suite.Require().NoError(err)
+
+	_, err = suite.db.ExecContext(ctx, "CREATE DATABASE test")
 	suite.Require().NoError(err)
 }
 
@@ -330,6 +334,9 @@ func (suite *MigratorTestSuite) TestFoo() {
 	_, err := suite.db.ExecContext(ctx, "CREATE TABLE test.testing (id INT PRIMARY KEY, name VARCHAR(64))")
 	suite.Require().NoError(err)
 
+	_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing(id, name) VALUES (1, 'mona')")
+	suite.Require().NoError(err)
+
 	connectionConfig, err := GetConnectionConfig(ctx, suite.mysqlContainer)
 	suite.Require().NoError(err)
 
@@ -338,18 +345,20 @@ func (suite *MigratorTestSuite) TestFoo() {
 	migrationContext.ApplierConnectionConfig = connectionConfig
 	migrationContext.InspectorConnectionConfig = connectionConfig
 	migrationContext.DatabaseName = "test"
-	migrationContext.SkipPortValidation = true
 	migrationContext.OriginalTableName = "testing"
 	migrationContext.SetConnectionConfig("innodb")
-	migrationContext.AlterStatementOptions = "ADD COLUMN foobar varchar(255), ENGINE=InnoDB"
+	migrationContext.AlterStatementOptions = "ADD COLUMN foobar varchar(255)"
 	migrationContext.ReplicaServerId = 99999
 	migrationContext.HeartbeatIntervalMilliseconds = 100
 	migrationContext.ThrottleHTTPIntervalMillis = 100
-	migrationContext.ThrottleHTTPTimeoutMillis = 1000
+	migrationContext.DMLBatchSize = 10
+	migrationContext.NumWorkers = 4
+	migrationContext.SkipPortValidation = true
 
 	//nolint:dogsled
 	_, filename, _, _ := runtime.Caller(0)
 	migrationContext.ServeSocketFile = filepath.Join(filepath.Dir(filename), "../../tmp/gh-ost.sock")
+	_ = os.Remove(filename)
 
 	migrator := NewMigrator(migrationContext, "0.0.0")
 
diff --git a/go/logic/server.go b/go/logic/server.go
index 4e41fd26b..616fa3534 100644
--- a/go/logic/server.go
+++ b/go/logic/server.go
@@ -32,6 +32,7 @@ var (
 )
 
 type printStatusFunc func(PrintStatusRule, io.Writer)
+type printWorkersFunc func(io.Writer)
 
 // Server listens for requests on a socket file or via TCP
 type Server struct {
@@ -40,14 +41,16 @@ type Server struct {
 	tcpListener      net.Listener
 	hooksExecutor    *HooksExecutor
 	printStatus      printStatusFunc
+	printWorkers     printWorkersFunc
 	isCPUProfiling   int64
 }
 
-func NewServer(migrationContext *base.MigrationContext, hooksExecutor *HooksExecutor, printStatus printStatusFunc) *Server {
+func NewServer(migrationContext *base.MigrationContext, hooksExecutor *HooksExecutor, printStatus printStatusFunc, printWorkers printWorkersFunc) *Server {
 	return &Server{
 		migrationContext: migrationContext,
 		hooksExecutor:    hooksExecutor,
 		printStatus:      printStatus,
+		printWorkers:     printWorkers,
 	}
 }
 
@@ -232,6 +235,9 @@ help                                 # This message
 		return ForcePrintStatusOnlyRule, nil
 	case "info", "status":
 		return ForcePrintStatusAndHintRule, nil
+	case "worker-stats":
+		this.printWorkers(writer)
+		return NoPrintStatusRule, nil
 	case "cpu-profile":
 		cpuProfile, err := this.runCPUProfile(arg)
 		if err == nil {
diff --git a/go/logic/streamer.go b/go/logic/streamer.go
deleted file mode 100644
index 338765256..000000000
--- a/go/logic/streamer.go
+++ /dev/null
@@ -1,219 +0,0 @@
-/*
-   Copyright 2022 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package logic
-
-import (
-	gosql "database/sql"
-	"fmt"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/github/gh-ost/go/base"
-	"github.com/github/gh-ost/go/binlog"
-	"github.com/github/gh-ost/go/mysql"
-
-	"github.com/openark/golib/sqlutils"
-)
-
-type BinlogEventListener struct {
-	async        bool
-	databaseName string
-	tableName    string
-	onDmlEvent   func(event *binlog.BinlogDMLEvent) error
-}
-
-const (
-	EventsChannelBufferSize       = 1
-	ReconnectStreamerSleepSeconds = 5
-)
-
-// EventsStreamer reads data from binary logs and streams it on. It acts as a publisher,
-// and interested parties may subscribe for per-table events.
-type EventsStreamer struct {
-	connectionConfig         *mysql.ConnectionConfig
-	db                       *gosql.DB
-	migrationContext         *base.MigrationContext
-	initialBinlogCoordinates *mysql.BinlogCoordinates
-	listeners                [](*BinlogEventListener)
-	listenersMutex           *sync.Mutex
-	eventsChannel            chan *binlog.BinlogEntry
-	binlogReader             *binlog.GoMySQLReader
-	name                     string
-}
-
-func NewEventsStreamer(migrationContext *base.MigrationContext) *EventsStreamer {
-	return &EventsStreamer{
-		connectionConfig: migrationContext.InspectorConnectionConfig,
-		migrationContext: migrationContext,
-		listeners:        [](*BinlogEventListener){},
-		listenersMutex:   &sync.Mutex{},
-		eventsChannel:    make(chan *binlog.BinlogEntry, EventsChannelBufferSize),
-		name:             "streamer",
-	}
-}
-
-// AddListener registers a new listener for binlog events, on a per-table basis
-func (this *EventsStreamer) AddListener(
-	async bool, databaseName string, tableName string, onDmlEvent func(event *binlog.BinlogDMLEvent) error) (err error) {
-	this.listenersMutex.Lock()
-	defer this.listenersMutex.Unlock()
-
-	if databaseName == "" {
-		return fmt.Errorf("Empty database name in AddListener")
-	}
-	if tableName == "" {
-		return fmt.Errorf("Empty table name in AddListener")
-	}
-	listener := &BinlogEventListener{
-		async:        async,
-		databaseName: databaseName,
-		tableName:    tableName,
-		onDmlEvent:   onDmlEvent,
-	}
-	this.listeners = append(this.listeners, listener)
-	return nil
-}
-
-// notifyListeners will notify relevant listeners with given DML event. Only
-// listeners registered for changes on the table on which the DML operates are notified.
-func (this *EventsStreamer) notifyListeners(binlogEvent *binlog.BinlogDMLEvent) {
-	this.listenersMutex.Lock()
-	defer this.listenersMutex.Unlock()
-
-	for _, listener := range this.listeners {
-		listener := listener
-		if !strings.EqualFold(listener.databaseName, binlogEvent.DatabaseName) {
-			continue
-		}
-		if !strings.EqualFold(listener.tableName, binlogEvent.TableName) {
-			continue
-		}
-		if listener.async {
-			go func() {
-				listener.onDmlEvent(binlogEvent)
-			}()
-		} else {
-			listener.onDmlEvent(binlogEvent)
-		}
-	}
-}
-
-func (this *EventsStreamer) InitDBConnections() (err error) {
-	EventsStreamerUri := this.connectionConfig.GetDBUri(this.migrationContext.DatabaseName)
-	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, EventsStreamerUri); err != nil {
-		return err
-	}
-	if _, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext, this.name); err != nil {
-		return err
-	}
-	if err := this.readCurrentBinlogCoordinates(); err != nil {
-		return err
-	}
-	if err := this.initBinlogReader(this.initialBinlogCoordinates); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// initBinlogReader creates and connects the reader: we hook up to a MySQL server as a replica
-func (this *EventsStreamer) initBinlogReader(binlogCoordinates *mysql.BinlogCoordinates) error {
-	goMySQLReader := binlog.NewGoMySQLReader(this.migrationContext)
-	if err := goMySQLReader.ConnectBinlogStreamer(*binlogCoordinates); err != nil {
-		return err
-	}
-	this.binlogReader = goMySQLReader
-	return nil
-}
-
-func (this *EventsStreamer) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinates {
-	return this.binlogReader.GetCurrentBinlogCoordinates()
-}
-
-func (this *EventsStreamer) GetReconnectBinlogCoordinates() *mysql.BinlogCoordinates {
-	return &mysql.BinlogCoordinates{LogFile: this.GetCurrentBinlogCoordinates().LogFile, LogPos: 4}
-}
-
-// readCurrentBinlogCoordinates reads master status from hooked server
-func (this *EventsStreamer) readCurrentBinlogCoordinates() error {
-	query := `show /* gh-ost readCurrentBinlogCoordinates */ master status`
-	foundMasterStatus := false
-	err := sqlutils.QueryRowsMap(this.db, query, func(m sqlutils.RowMap) error {
-		this.initialBinlogCoordinates = &mysql.BinlogCoordinates{
-			LogFile: m.GetString("File"),
-			LogPos:  m.GetInt64("Position"),
-		}
-		foundMasterStatus = true
-
-		return nil
-	})
-	if err != nil {
-		return err
-	}
-	if !foundMasterStatus {
-		return fmt.Errorf("Got no results from SHOW MASTER STATUS. Bailing out")
-	}
-	this.migrationContext.Log.Debugf("Streamer binlog coordinates: %+v", *this.initialBinlogCoordinates)
-	return nil
-}
-
-// StreamEvents will begin streaming events. It will be blocking, so should be
-// executed by a goroutine
-func (this *EventsStreamer) StreamEvents(canStopStreaming func() bool) error {
-	go func() {
-		for binlogEntry := range this.eventsChannel {
-			if binlogEntry.DmlEvent != nil {
-				this.notifyListeners(binlogEntry.DmlEvent)
-			}
-		}
-	}()
-	// The next should block and execute forever, unless there's a serious error
-	var successiveFailures int64
-	var lastAppliedRowsEventHint mysql.BinlogCoordinates
-	for {
-		if canStopStreaming() {
-			return nil
-		}
-		if err := this.binlogReader.StreamEvents(canStopStreaming, this.eventsChannel); err != nil {
-			if canStopStreaming() {
-				return nil
-			}
-
-			this.migrationContext.Log.Infof("StreamEvents encountered unexpected error: %+v", err)
-			this.migrationContext.MarkPointOfInterest()
-			time.Sleep(ReconnectStreamerSleepSeconds * time.Second)
-
-			// See if there's retry overflow
-			if this.binlogReader.LastAppliedRowsEventHint.Equals(&lastAppliedRowsEventHint) {
-				successiveFailures += 1
-			} else {
-				successiveFailures = 0
-			}
-			if successiveFailures >= this.migrationContext.MaxRetries() {
-				return fmt.Errorf("%d successive failures in streamer reconnect at coordinates %+v", successiveFailures, this.GetReconnectBinlogCoordinates())
-			}
-
-			// Reposition at same binlog file.
-			lastAppliedRowsEventHint = this.binlogReader.LastAppliedRowsEventHint
-			this.migrationContext.Log.Infof("Reconnecting... Will resume at %+v", lastAppliedRowsEventHint)
-			if err := this.initBinlogReader(this.GetReconnectBinlogCoordinates()); err != nil {
-				return err
-			}
-			this.binlogReader.LastAppliedRowsEventHint = lastAppliedRowsEventHint
-		}
-	}
-}
-
-func (this *EventsStreamer) Close() (err error) {
-	err = this.binlogReader.Close()
-	this.migrationContext.Log.Infof("Closed streamer connection. err=%+v", err)
-	return err
-}
-
-func (this *EventsStreamer) Teardown() {
-	this.db.Close()
-}
diff --git a/go/logic/streamer_test.go b/go/logic/streamer_test.go
index 301074e3f..756257b6d 100644
--- a/go/logic/streamer_test.go
+++ b/go/logic/streamer_test.go
@@ -1,282 +1,282 @@
 package logic
 
-import (
-	"context"
-	"database/sql"
-	gosql "database/sql"
-	"fmt"
-	"testing"
-	"time"
+// import (
+// 	"context"
+// 	"database/sql"
+// 	gosql "database/sql"
+// 	"fmt"
+// 	"testing"
+// 	"time"
 
-	"github.com/github/gh-ost/go/base"
-	"github.com/github/gh-ost/go/binlog"
-	"github.com/stretchr/testify/suite"
-	"github.com/testcontainers/testcontainers-go"
-	"github.com/testcontainers/testcontainers-go/wait"
+// 	"github.com/github/gh-ost/go/base"
+// 	"github.com/github/gh-ost/go/binlog"
+// 	"github.com/stretchr/testify/suite"
+// 	"github.com/testcontainers/testcontainers-go"
+// 	"github.com/testcontainers/testcontainers-go/wait"
 
-	"golang.org/x/sync/errgroup"
-)
-
-type EventsStreamerTestSuite struct {
-	suite.Suite
-
-	mysqlContainer testcontainers.Container
-	db             *gosql.DB
-}
-
-func (suite *EventsStreamerTestSuite) SetupSuite() {
-	ctx := context.Background()
-	req := testcontainers.ContainerRequest{
-		Image:        "mysql:8.0.40",
-		Env:          map[string]string{"MYSQL_ROOT_PASSWORD": "root-password"},
-		ExposedPorts: []string{"3306/tcp"},
-		WaitingFor:   wait.ForListeningPort("3306/tcp"),
-	}
-
-	mysqlContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
-		ContainerRequest: req,
-		Started:          true,
-	})
-	suite.Require().NoError(err)
-
-	suite.mysqlContainer = mysqlContainer
-
-	dsn, err := GetDSN(ctx, mysqlContainer)
-	suite.Require().NoError(err)
-
-	db, err := gosql.Open("mysql", dsn)
-	suite.Require().NoError(err)
-
-	suite.db = db
-}
-
-func (suite *EventsStreamerTestSuite) TeardownSuite() {
-	ctx := context.Background()
-
-	suite.Assert().NoError(suite.db.Close())
-	suite.Assert().NoError(suite.mysqlContainer.Terminate(ctx))
-}
-
-func (suite *EventsStreamerTestSuite) SetupTest() {
-	ctx := context.Background()
-
-	_, err := suite.db.ExecContext(ctx, "CREATE DATABASE test")
-	suite.Require().NoError(err)
-}
-
-func (suite *EventsStreamerTestSuite) TearDownTest() {
-	ctx := context.Background()
-
-	_, err := suite.db.ExecContext(ctx, "DROP DATABASE test")
-	suite.Require().NoError(err)
-}
+// 	"golang.org/x/sync/errgroup"
+// )
+
+// type EventsStreamerTestSuite struct {
+// 	suite.Suite
+
+// 	mysqlContainer testcontainers.Container
+// 	db             *gosql.DB
+// }
+
+// func (suite *EventsStreamerTestSuite) SetupSuite() {
+// 	ctx := context.Background()
+// 	req := testcontainers.ContainerRequest{
+// 		Image:        "mysql:8.0.40",
+// 		Env:          map[string]string{"MYSQL_ROOT_PASSWORD": "root-password"},
+// 		ExposedPorts: []string{"3306/tcp"},
+// 		WaitingFor:   wait.ForListeningPort("3306/tcp"),
+// 	}
+
+// 	mysqlContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
+// 		ContainerRequest: req,
+// 		Started:          true,
+// 	})
+// 	suite.Require().NoError(err)
+
+// 	suite.mysqlContainer = mysqlContainer
+
+// 	dsn, err := GetDSN(ctx, mysqlContainer)
+// 	suite.Require().NoError(err)
+
+// 	db, err := gosql.Open("mysql", dsn)
+// 	suite.Require().NoError(err)
+
+// 	suite.db = db
+// }
+
+// func (suite *EventsStreamerTestSuite) TeardownSuite() {
+// 	ctx := context.Background()
+
+// 	suite.Assert().NoError(suite.db.Close())
+// 	suite.Assert().NoError(suite.mysqlContainer.Terminate(ctx))
+// }
+
+// func (suite *EventsStreamerTestSuite) SetupTest() {
+// 	ctx := context.Background()
+
+// 	_, err := suite.db.ExecContext(ctx, "CREATE DATABASE test")
+// 	suite.Require().NoError(err)
+// }
+
+// func (suite *EventsStreamerTestSuite) TearDownTest() {
+// 	ctx := context.Background()
+
+// 	_, err := suite.db.ExecContext(ctx, "DROP DATABASE test")
+// 	suite.Require().NoError(err)
+// }
 
-func (suite *EventsStreamerTestSuite) TestStreamEvents() {
-	ctx := context.Background()
+// func (suite *EventsStreamerTestSuite) TestStreamEvents() {
+// 	ctx := context.Background()
 
-	_, err := suite.db.ExecContext(ctx, "CREATE TABLE test.testing (id INT PRIMARY KEY, name VARCHAR(255))")
-	suite.Require().NoError(err)
+// 	_, err := suite.db.ExecContext(ctx, "CREATE TABLE test.testing (id INT PRIMARY KEY, name VARCHAR(255))")
+// 	suite.Require().NoError(err)
 
-	connectionConfig, err := GetConnectionConfig(ctx, suite.mysqlContainer)
-	suite.Require().NoError(err)
+// 	connectionConfig, err := GetConnectionConfig(ctx, suite.mysqlContainer)
+// 	suite.Require().NoError(err)
 
-	migrationContext := base.NewMigrationContext()
-	migrationContext.ApplierConnectionConfig = connectionConfig
-	migrationContext.InspectorConnectionConfig = connectionConfig
-	migrationContext.DatabaseName = "test"
-	migrationContext.SkipPortValidation = true
-	migrationContext.ReplicaServerId = 99999
+// 	migrationContext := base.NewMigrationContext()
+// 	migrationContext.ApplierConnectionConfig = connectionConfig
+// 	migrationContext.InspectorConnectionConfig = connectionConfig
+// 	migrationContext.DatabaseName = "test"
+// 	migrationContext.SkipPortValidation = true
+// 	migrationContext.ReplicaServerId = 99999
 
-	migrationContext.SetConnectionConfig("innodb")
+// 	migrationContext.SetConnectionConfig("innodb")
 
-	streamer := NewEventsStreamer(migrationContext)
+// 	streamer := NewEventsStreamer(migrationContext)
 
-	err = streamer.InitDBConnections()
-	suite.Require().NoError(err)
-	defer streamer.Close()
-	defer streamer.Teardown()
-
-	streamCtx, cancel := context.WithCancel(context.Background())
-
-	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
-	err = streamer.AddListener(false, "test", "testing", func(event *binlog.BinlogDMLEvent) error {
-		dmlEvents = append(dmlEvents, event)
-
-		// Stop once we've collected three events
-		if len(dmlEvents) == 3 {
-			cancel()
-		}
+// 	err = streamer.InitDBConnections()
+// 	suite.Require().NoError(err)
+// 	defer streamer.Close()
+// 	defer streamer.Teardown()
+
+// 	streamCtx, cancel := context.WithCancel(context.Background())
+
+// 	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
+// 	err = streamer.AddListener(false, "test", "testing", func(event *binlog.BinlogDMLEvent) error {
+// 		dmlEvents = append(dmlEvents, event)
+
+// 		// Stop once we've collected three events
+// 		if len(dmlEvents) == 3 {
+// 			cancel()
+// 		}
 
-		return nil
-	})
-	suite.Require().NoError(err)
+// 		return nil
+// 	})
+// 	suite.Require().NoError(err)
 
-	group := errgroup.Group{}
-	group.Go(func() error {
-		//nolint:contextcheck
-		return streamer.StreamEvents(func() bool {
-			return streamCtx.Err() != nil
-		})
-	})
+// 	group := errgroup.Group{}
+// 	group.Go(func() error {
+// 		//nolint:contextcheck
+// 		return streamer.StreamEvents(func() bool {
+// 			return streamCtx.Err() != nil
+// 		})
+// 	})
 
-	group.Go(func() error {
-		var err error
-
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (1, 'foo')")
-		if err != nil {
-			return err
-		}
-
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (2, 'bar')")
-		if err != nil {
-			return err
-		}
+// 	group.Go(func() error {
+// 		var err error
+
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (1, 'foo')")
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (2, 'bar')")
+// 		if err != nil {
+// 			return err
+// 		}
 
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (3, 'baz')")
-		if err != nil {
-			return err
-		}
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (3, 'baz')")
+// 		if err != nil {
+// 			return err
+// 		}
 
-		// Bug: Need to write fourth event to hit the canStopStreaming function again
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (4, 'qux')")
-		if err != nil {
-			return err
-		}
+// 		// Bug: Need to write fourth event to hit the canStopStreaming function again
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (4, 'qux')")
+// 		if err != nil {
+// 			return err
+// 		}
 
-		return nil
-	})
-
-	err = group.Wait()
-	suite.Require().NoError(err)
-
-	suite.Require().Len(dmlEvents, 3)
-}
-
-func (suite *EventsStreamerTestSuite) TestStreamEventsAutomaticallyReconnects() {
-	ctx := context.Background()
-
-	_, err := suite.db.ExecContext(ctx, "CREATE TABLE test.testing (id INT PRIMARY KEY, name VARCHAR(255))")
-	suite.Require().NoError(err)
-
-	connectionConfig, err := GetConnectionConfig(ctx, suite.mysqlContainer)
-	suite.Require().NoError(err)
-
-	migrationContext := base.NewMigrationContext()
-	migrationContext.ApplierConnectionConfig = connectionConfig
-	migrationContext.InspectorConnectionConfig = connectionConfig
-	migrationContext.DatabaseName = "test"
-	migrationContext.SkipPortValidation = true
-	migrationContext.ReplicaServerId = 99999
-
-	migrationContext.SetConnectionConfig("innodb")
-
-	streamer := NewEventsStreamer(migrationContext)
-
-	err = streamer.InitDBConnections()
-	suite.Require().NoError(err)
-	defer streamer.Close()
-	defer streamer.Teardown()
-
-	streamCtx, cancel := context.WithCancel(context.Background())
-
-	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
-	err = streamer.AddListener(false, "test", "testing", func(event *binlog.BinlogDMLEvent) error {
-		dmlEvents = append(dmlEvents, event)
-
-		// Stop once we've collected three events
-		if len(dmlEvents) == 3 {
-			cancel()
-		}
-
-		return nil
-	})
-	suite.Require().NoError(err)
-
-	group := errgroup.Group{}
-	group.Go(func() error {
-		//nolint:contextcheck
-		return streamer.StreamEvents(func() bool {
-			return streamCtx.Err() != nil
-		})
-	})
-
-	group.Go(func() error {
-		var err error
-
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (1, 'foo')")
-		if err != nil {
-			return err
-		}
-
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (2, 'bar')")
-		if err != nil {
-			return err
-		}
-
-		var currentConnectionId int
-		err = suite.db.QueryRowContext(ctx, "SELECT CONNECTION_ID()").Scan(&currentConnectionId)
-		if err != nil {
-			return err
-		}
-
-		//nolint:execinquery
-		rows, err := suite.db.Query("SHOW FULL PROCESSLIST")
-		if err != nil {
-			return err
-		}
-		defer rows.Close()
-
-		connectionIdsToKill := make([]int, 0)
-
-		var id, stateTime int
-		var user, host, dbName, command, state, info sql.NullString
-		for rows.Next() {
-			err = rows.Scan(&id, &user, &host, &dbName, &command, &stateTime, &state, &info)
-			if err != nil {
-				return err
-			}
-
-			fmt.Printf("id: %d, user: %s, host: %s, dbName: %s, command: %s, time: %d, state: %s, info: %s\n", id, user.String, host.String, dbName.String, command.String, stateTime, state.String, info.String)
-
-			if id != currentConnectionId && user.String == "root" {
-				connectionIdsToKill = append(connectionIdsToKill, id)
-			}
-		}
-
-		if err := rows.Err(); err != nil {
-			return err
-		}
-
-		for _, connectionIdToKill := range connectionIdsToKill {
-			_, err = suite.db.ExecContext(ctx, "KILL ?", connectionIdToKill)
-			if err != nil {
-				return err
-			}
-		}
-
-		// Bug: We need to wait here for the streamer to reconnect
-		time.Sleep(time.Second * 2)
-
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (3, 'baz')")
-		if err != nil {
-			return err
-		}
-
-		// Bug: Need to write fourth event to hit the canStopStreaming function again
-		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (4, 'qux')")
-		if err != nil {
-			return err
-		}
-
-		return nil
-	})
-
-	err = group.Wait()
-	suite.Require().NoError(err)
-
-	suite.Require().Len(dmlEvents, 3)
-}
-
-func TestEventsStreamer(t *testing.T) {
-	suite.Run(t, new(EventsStreamerTestSuite))
-}
+// 		return nil
+// 	})
+
+// 	err = group.Wait()
+// 	suite.Require().NoError(err)
+
+// 	suite.Require().Len(dmlEvents, 3)
+// }
+
+// func (suite *EventsStreamerTestSuite) TestStreamEventsAutomaticallyReconnects() {
+// 	ctx := context.Background()
+
+// 	_, err := suite.db.ExecContext(ctx, "CREATE TABLE test.testing (id INT PRIMARY KEY, name VARCHAR(255))")
+// 	suite.Require().NoError(err)
+
+// 	connectionConfig, err := GetConnectionConfig(ctx, suite.mysqlContainer)
+// 	suite.Require().NoError(err)
+
+// 	migrationContext := base.NewMigrationContext()
+// 	migrationContext.ApplierConnectionConfig = connectionConfig
+// 	migrationContext.InspectorConnectionConfig = connectionConfig
+// 	migrationContext.DatabaseName = "test"
+// 	migrationContext.SkipPortValidation = true
+// 	migrationContext.ReplicaServerId = 99999
+
+// 	migrationContext.SetConnectionConfig("innodb")
+
+// 	streamer := NewEventsStreamer(migrationContext)
+
+// 	err = streamer.InitDBConnections()
+// 	suite.Require().NoError(err)
+// 	defer streamer.Close()
+// 	defer streamer.Teardown()
+
+// 	streamCtx, cancel := context.WithCancel(context.Background())
+
+// 	dmlEvents := make([]*binlog.BinlogDMLEvent, 0)
+// 	err = streamer.AddListener(false, "test", "testing", func(event *binlog.BinlogDMLEvent) error {
+// 		dmlEvents = append(dmlEvents, event)
+
+// 		// Stop once we've collected three events
+// 		if len(dmlEvents) == 3 {
+// 			cancel()
+// 		}
+
+// 		return nil
+// 	})
+// 	suite.Require().NoError(err)
+
+// 	group := errgroup.Group{}
+// 	group.Go(func() error {
+// 		//nolint:contextcheck
+// 		return streamer.StreamEvents(func() bool {
+// 			return streamCtx.Err() != nil
+// 		})
+// 	})
+
+// 	group.Go(func() error {
+// 		var err error
+
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (1, 'foo')")
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (2, 'bar')")
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		var currentConnectionId int
+// 		err = suite.db.QueryRowContext(ctx, "SELECT CONNECTION_ID()").Scan(&currentConnectionId)
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		//nolint:execinquery
+// 		rows, err := suite.db.Query("SHOW FULL PROCESSLIST")
+// 		if err != nil {
+// 			return err
+// 		}
+// 		defer rows.Close()
+
+// 		connectionIdsToKill := make([]int, 0)
+
+// 		var id, stateTime int
+// 		var user, host, dbName, command, state, info sql.NullString
+// 		for rows.Next() {
+// 			err = rows.Scan(&id, &user, &host, &dbName, &command, &stateTime, &state, &info)
+// 			if err != nil {
+// 				return err
+// 			}
+
+// 			fmt.Printf("id: %d, user: %s, host: %s, dbName: %s, command: %s, time: %d, state: %s, info: %s\n", id, user.String, host.String, dbName.String, command.String, stateTime, state.String, info.String)
+
+// 			if id != currentConnectionId && user.String == "root" {
+// 				connectionIdsToKill = append(connectionIdsToKill, id)
+// 			}
+// 		}
+
+// 		if err := rows.Err(); err != nil {
+// 			return err
+// 		}
+
+// 		for _, connectionIdToKill := range connectionIdsToKill {
+// 			_, err = suite.db.ExecContext(ctx, "KILL ?", connectionIdToKill)
+// 			if err != nil {
+// 				return err
+// 			}
+// 		}
+
+// 		// Bug: We need to wait here for the streamer to reconnect
+// 		time.Sleep(time.Second * 2)
+
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (3, 'baz')")
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		// Bug: Need to write fourth event to hit the canStopStreaming function again
+// 		_, err = suite.db.ExecContext(ctx, "INSERT INTO test.testing (id, name) VALUES (4, 'qux')")
+// 		if err != nil {
+// 			return err
+// 		}
+
+// 		return nil
+// 	})
+
+// 	err = group.Wait()
+// 	suite.Require().NoError(err)
+
+// 	suite.Require().Len(dmlEvents, 3)
+// }
+
+// func TestEventsStreamer(t *testing.T) {
+// 	suite.Run(t, new(EventsStreamerTestSuite))
+// }
diff --git a/localtests/test.sh b/localtests/test.sh
index 0c8670cb6..9297953a9 100755
--- a/localtests/test.sh
+++ b/localtests/test.sh
@@ -24,6 +24,7 @@ master_port=
 replica_host=
 replica_port=
 original_sql_mode=
+docker=false
 
 OPTIND=1
 while getopts "b:s:d" OPTION
diff --git a/script/test b/script/test
index 5c32b370c..bec4f67f5 100755
--- a/script/test
+++ b/script/test
@@ -5,7 +5,7 @@ set -e
 . script/bootstrap
 
 echo "Verifying code is formatted via 'gofmt -s -w  go/'"
-gofmt -s -w  go/
+gofmt -s -w go/
 git diff --exit-code --quiet
 
 echo "Building"
@@ -14,4 +14,4 @@ script/build
 cd .gopath/src/github.com/github/gh-ost
 
 echo "Running unit tests"
-go test -v -covermode=atomic ./go/...
+go test -v -p 1 -covermode=atomic ./go/...