fix repeated kafka run

a bug introduced in 7406c29.

Author: ffffwh

drivers/mysql/driver.go

@@ -495,7 +495,6 @@ func (d *Driver) StartTask(cfg *drivers.TaskConfig) (*drivers.TaskHandle, *drive
 				d.logger.Debug("found kafka", "KafkaConfig", driverConfig.KafkaConfig)
 				h.runner = kafka.NewKafkaRunner(ctx, driverConfig.KafkaConfig, d.logger,
 					d.storeManager, d.config.NatsAdvertise, h.waitCh)
-				go h.runner.Run()
 			} else {
 				h.runner, err = mysql.NewApplier(ctx, driverConfig, d.logger, d.storeManager,
 					d.config.NatsAdvertise, h.waitCh,d.eventer, h.taskConfig)

drivers/mysql/kafka/kafka3.go

@@ -352,6 +352,7 @@ func (kr *KafkaRunner) handleFullCopy() {
 	}
 }
 func (kr *KafkaRunner) handleIncr() {
+	kr.logger.Debug("handleIncr")
 	var err error
 	bigEntries := &common.BinlogEntries{}
 	for !kr.shutdown {
@@ -696,6 +697,7 @@ func (kr *KafkaRunner) kafkaTransformSnapshotData(
 }
 
 func (kr *KafkaRunner) kafkaTransformDMLEventQuery(dmlEvent *common.BinlogEntry) (err error) {
+	kr.logger.Debug("kafkaTransformDMLEventQuery", "gno", dmlEvent.Coordinates.GNO)
 	txSid := dmlEvent.Coordinates.GetSid()
 
 	for i, _ := range dmlEvent.Events {