Fixed race in ddl commands cleanup (#481)

* Fixed race in ddl commands cleanup

* Call cleanup on error

cluster/dragon/logadaptor/logrus_adaptor.go

@@ -0,0 +1,54 @@
+package logadaptor
+
+import (
+	"github.com/lni/dragonboat/v3/logger"
+	log "github.com/sirupsen/logrus"
+)
+
+/*
+This adaptor allows us to plug the dragonboat logging into the logrus logger we use in Prana.
+*/
+
+func init() {
+	logger.SetLoggerFactory(logrusLogFactory)
+}
+
+func logrusLogFactory(pkgName string) logger.ILogger {
+	return &LogrusILogger{}
+}
+
+type LogrusILogger struct {
+	level logger.LogLevel
+}
+
+func (l *LogrusILogger) SetLevel(level logger.LogLevel) {
+	l.level = level
+}
+
+func (l *LogrusILogger) Debugf(format string, args ...interface{}) {
+	if l.level >= logger.DEBUG {
+		log.Debugf(format, args...)
+	}
+}
+
+func (l *LogrusILogger) Infof(format string, args ...interface{}) {
+	if l.level >= logger.INFO {
+		log.Infof(format, args...)
+	}
+}
+
+func (l *LogrusILogger) Warningf(format string, args ...interface{}) {
+	if l.level >= logger.WARNING {
+		log.Warnf(format, args...)
+	}
+}
+
+func (l *LogrusILogger) Errorf(format string, args ...interface{}) {
+	if l.level >= logger.ERROR {
+		log.Errorf(format, args...)
+	}
+}
+
+func (l *LogrusILogger) Panicf(format string, args ...interface{}) {
+	log.Fatalf(format, args...)
+}

command/create_index_command.go

@@ -148,6 +148,8 @@ func (c *CreateIndexCommand) AfterPhase(phase int32) error {
 }
 
 func (c *CreateIndexCommand) Cleanup() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if c.indexInfo == nil {
 		return
 	}

command/create_mv_command.go

@@ -218,6 +218,8 @@ func (c *CreateMVCommand) AfterPhase(phase int32) error {
 }
 
 func (c *CreateMVCommand) Cleanup() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if c.mv == nil {
 		return
 	}

command/create_source_command.go

@@ -211,6 +211,8 @@ func (c *CreateSourceCommand) AfterPhase(phase int32) error {
 }
 
 func (c *CreateSourceCommand) Cleanup() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if c.sourceInfo == nil {
 		return
 	}

command/ddl_runner.go

@@ -216,6 +216,9 @@ func (d *DDLCommandRunner) HandleDdlMessage(notification remoting.ClusterMessage
 		return nil
 	}
 	err := com.OnPhase(phase)
+	if err != nil {
+		com.Cleanup()
+	}
 	if phase == int32(com.NumPhases()-1) {
 		// Final phase so delete the command
 		d.commands.Delete(skey)