misc: cleanup and logs

Author: ffffwh

drivers/mysql/kafka/kafka2.go

@@ -93,7 +93,7 @@ func (k *KafkaManager) SendMessages(logger hclog.Logger, topics []string, keys [
 	}
 	if errs := k.producer.SendMessages(msgs); errs != nil {
 		for _, err := range errs.(sarama.ProducerErrors) {
-			logger.Error("send messages to kafka failed: ", err)
+			logger.Error("SendMessages failed", "err", err)
 		}
 		return errs
 	}

drivers/mysql/kafka/kafka3.go

@@ -1043,7 +1043,6 @@ func (kr *KafkaRunner) kafkaTransformDMLEventQueries(dmlEntries []*common.Binlog
 			tableIdents = append(tableIdents, tableIdent)
 			keysBs = append(keysBs, kBs)
 			valuesBs = append(valuesBs, vBs)
-			vBs = []byte(strings.Replace(string(vBs), "\"field\":\"snapshot\"", "\"default\":false,\"field\":\"snapshot\"", -1))
 
 			// tombstone event for DELETE
 			if dataEvent.DML == common.DeleteDML {

drivers/mysql/mysql/base/binlog.go

@@ -57,7 +57,6 @@ func (m *GtidItemMap) GetItem(u uuid.UUID) (item *GtidItem) {
 	}
 }
 func GetIntervals(set *gomysql.MysqlGTIDSet, uuidStr string) gomysql.IntervalSlice {
-	// TODO read lock
 	item, ok := set.Sets[uuidStr]
 	if ok {
 		return item.Intervals

drivers/mysql/mysql/dumper.go

@@ -206,7 +206,7 @@ func (d *dumper) getChunkData() (nRows int64, err error) {
 			select {
 			case d.resultsChannel <- entry:
 				atomic.AddInt64(d.memory, int64(entry.Size()))
-				d.logger.Debug("*** memory", "memory", atomic.LoadInt64(d.memory))
+				//d.logger.Debug("*** memory", "memory", atomic.LoadInt64(d.memory))
 				if !timer.Stop() {
 					<-timer.C
 				}