Close remaining DB connections

go/logic/applier.go

@@ -291,6 +291,8 @@ func (this *Applier) WriteChangelogState(value string) (string, error) {
 }
 
 func (this *Applier) FinalCleanup() {
+	this.db.Close()
+	this.singletonDB.Close()
 	this.finishedMigrating = true
 }
 

go/logic/migrator.go

@@ -1224,6 +1224,7 @@ func (this *Migrator) finalCleanup() error {
 
 	this.finishedMigrating = true
 	this.applier.FinalCleanup()
+	this.eventsStreamer.FinalCleanup()
 
 	return nil
 }

go/logic/streamer.go

@@ -192,7 +192,14 @@ func (this *EventsStreamer) StreamEvents(canStopStreaming func() bool) 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
+			}
+
 			log.Infof("StreamEvents encountered unexpected error: %+v", err)
 			this.migrationContext.MarkPointOfInterest()
 			time.Sleep(ReconnectStreamerSleepSeconds * time.Second)
@@ -223,3 +230,8 @@ func (this *EventsStreamer) Close() (err error) {
 	log.Infof("Closed streamer connection. err=%+v", err)
 	return err
 }
+
+func (this *EventsStreamer) FinalCleanup() {
+	this.db.Close()
+	return
+}

go/mysql/utils.go

@@ -51,6 +51,8 @@ func GetReplicationLag(connectionConfig *ConnectionConfig) (replicationLag time.
 		replicationLag = time.Duration(secondsBehindMaster.Int64) * time.Second
 		return nil
 	})
+
+	db.Close()
 	return replicationLag, err
 }