Merge pull request #295 from brokercap/v2.3.11

V2.3.11

Author: jc3wish

README.MD

@@ -124,11 +124,11 @@ make install prefix=./target
 ##### 二进制文件安装
 `````sh
 
-wget https://github.com/brokercap/Bifrost/releases/download/v2.3.6-beta/bifrost_v2.3.6-beta_Linux-amd64-bin.tar.gz
+wget https://github.com/brokercap/Bifrost/releases/download/v2.3.11-beta/bifrost_v2.3.11-beta_Linux-amd64-bin.tar.gz
 
-tar -zxvf bifrost_v2.3.6-beta_Linux-amd64-bin.tar.gz
+tar -zxvf bifrost_v2.3.11-beta_Linux-amd64-bin.tar.gz
 
-cd bifrost_v2.3.6-beta_Linux-amd64-bin/bin && chmod a+x ./Bifrost*
+cd bifrost_v2.3.11-beta_Linux-amd64-bin/bin && chmod a+x ./Bifrost*
 
 `````
 

changelog.txt

@@ -1,3 +1,14 @@
+v2.3.11-beta   2025-03-07
+1. plugin mysql 修复获取表元数据的时候连接异常导致的连接错误没被识别的bug
+2. kafka plugin 消息统计不再使用计数器，而是直接使用切片的长度
+3. 修复开启了FilterUpdate，但是源表中存在复杂的json类型字段匹配错误，导致的异常退出bug
+
+v2.3.10-beta   2025-01-19
+1. fix mongo batch _id contains ' " ' bug
+   修改mongo源全量同步到目标库的时候，_id 字段包含双引号的bug
+2. Update the logic for determining whether it is Starrocks, supporting specifying it as Starrocks in the URI and forcing it to be set as Starrocks
+   支持uri连配配置中设置starrocks字样，会被识别为starrocks数据库
+
 v2.3.8-beta   2024-11-05
 1. fix mysql select tinyint(1) , value > 1 , transfer to bool ,result false bug
 

config/version.go

@@ -16,4 +16,4 @@ limitations under the License.
 
 package config
 
-const VERSION = "v2.3.8-beta"
+const VERSION = "v2.3.11-beta"

input/mongo/batch.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	outputDriver "github.com/brokercap/Bifrost/plugin/driver"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"log"
@@ -86,14 +87,17 @@ func (c *MongoInput) TableBatchStart(collection *mongo.Collection, perBatchLimit
 		if len(batchResult) == 0 {
 			break
 		}
+		nextMinId = batchResult[len(batchResult)-1]["_id"]
 		for _, batchInfo := range batchResult {
+			if docId, ok := batchInfo["_id"].(primitive.ObjectID); ok {
+				batchInfo["_id"] = docId.Hex()
+			}
 			eventData := c.BatchResult2RowEvent(schemaName, tableName, batchInfo)
 			c.callback(eventData)
 		}
 		if len(batchResult) < perBatchLimit {
 			break
 		}
-		nextMinId = batchResult[len(batchResult)-1]["_id"]
 	}
 	return nil
 }

plugin/kafka/src/kafka.go

@@ -57,7 +57,6 @@ type PluginParam struct {
 
 	dataList         []*sarama.ProducerMessage
 	commitBinlogList []*pluginDriver.PluginDataType
-	dataCurrentCount int
 }
 
 func NewConn() pluginDriver.Driver {
@@ -151,7 +150,6 @@ func (This *Conn) GetParam(p interface{}) (interface{}, error) {
 	if len(param.dataList) == 0 {
 		param.dataList = make([]*sarama.ProducerMessage, 0)
 		param.commitBinlogList = make([]*pluginDriver.PluginDataType, 0)
-		param.dataCurrentCount = 0
 	}
 	This.p = param
 	return param, nil
@@ -267,9 +265,8 @@ func (This *Conn) sendToList(data *pluginDriver.PluginDataType, retry bool, isCo
 					This.p.commitBinlogList[n0] = data
 				}
 			}
-			This.p.dataCurrentCount++
 		}
-		if This.p.dataCurrentCount >= This.p.BatchSize {
+		if len(This.p.dataList) >= This.p.BatchSize {
 			LastSuccessCommitData, err = This.sendToKafkaByBatch()
 		}
 	} else {
@@ -314,17 +311,16 @@ func (This *Conn) sendToKafkaByBatch() (*pluginDriver.PluginDataType, error) {
 			return nil, This.err
 		}
 	}
-	if This.p.dataCurrentCount == 0 {
+	if len(This.p.dataList) == 0 {
 		return nil, nil
 	}
 	var err error
 	var binlogEvent *pluginDriver.PluginDataType
-	if This.p.dataCurrentCount > This.p.BatchSize {
+	if len(This.p.dataList) > This.p.BatchSize {
 		list := This.p.dataList[:This.p.BatchSize]
 		err = This.producer.SendMessages(list)
 		if err == nil {
 			This.p.dataList = This.p.dataList[This.p.BatchSize:]
-			This.p.dataCurrentCount -= This.p.BatchSize
 			if len(This.p.commitBinlogList) > 0 {
 				binlogEvent = This.p.commitBinlogList[0]
 				This.p.commitBinlogList = This.p.commitBinlogList[1:]
@@ -334,7 +330,6 @@ func (This *Conn) sendToKafkaByBatch() (*pluginDriver.PluginDataType, error) {
 		err = This.producer.SendMessages(This.p.dataList)
 		if err == nil {
 			This.p.dataList = make([]*sarama.ProducerMessage, 0)
-			This.p.dataCurrentCount = 0
 			if len(This.p.commitBinlogList) > 0 {
 				binlogEvent = This.p.commitBinlogList[0]
 				This.p.commitBinlogList = This.p.commitBinlogList[1:]

plugin/mysql/src/schema.go

@@ -47,10 +47,10 @@ func (This *mysqlDB) Close() bool {
 }
 
 func (This *mysqlDB) GetSchemaList() (data []string) {
-	rows, err := This.conn.Query("SHOW DATABASES", []driver.Value{})
-	if err != nil {
-		log.Printf("[ERROR] output[%s] GetSchemaList err:%+v \n", OutputName, err)
-		This.err = err
+	var rows driver.Rows
+	rows, This.err = This.conn.Query("SHOW DATABASES", []driver.Value{})
+	if This.err != nil {
+		log.Printf("[ERROR] output[%s] GetSchemaList err:%+v \n", OutputName, This.err)
 		return
 	}
 	defer rows.Close()
@@ -120,13 +120,14 @@ func (This *mysqlDB) GetTableFields(schema, table string) (data []TableStruct) {
 	p := make([]driver.Value, 0)
 	p = append(p, schema)
 	p = append(p, table)
-	rows, err := This.conn.Query(sql, p)
-	if err != nil {
+	var rows driver.Rows
+	rows, This.err = This.conn.Query(sql, p)
+	if This.err != nil {
 		return
 	}
 	defer rows.Close()
-	if err != nil {
-		log.Printf("[ERROR] output[%s] GetTableFields schema:%s table:%s err:%+v \n", OutputName, schema, table, err)
+	if This.err != nil {
+		log.Printf("[ERROR] output[%s] GetTableFields schema:%s table:%s err:%+v \n", OutputName, schema, table, This.err)
 		return FieldList
 	}
 	for {
@@ -191,26 +192,27 @@ func (This *mysqlDB) GetTableFields(schema, table string) (data []TableStruct) {
 }
 
 func (This *mysqlDB) Begin() error {
-	_, err := This.conn.Exec("BEGIN", make([]driver.Value, 0))
-	return err
+	_, This.err = This.conn.Exec("BEGIN", make([]driver.Value, 0))
+	return This.err
 }
 
 func (This *mysqlDB) Commit() error {
-	_, err := This.conn.Exec("COMMIT", make([]driver.Value, 0))
-	return err
+	_, This.err = This.conn.Exec("COMMIT", make([]driver.Value, 0))
+	return This.err
 }
 
 func (This *mysqlDB) Rollback() error {
-	_, err := This.conn.Exec("ROLLBACK", make([]driver.Value, 0))
-	return err
+	_, This.err = This.conn.Exec("ROLLBACK", make([]driver.Value, 0))
+	return This.err
 }
 
 func (This *mysqlDB) ShowTableCreate(schema, table string) string {
 	sql := "SHOW CREATE TABLE `" + schema + "`.`" + table + "`"
 	p := make([]driver.Value, 0)
-	rows, err := This.conn.Query(sql, p)
-	if err != nil {
-		log.Printf("[ERROR] output[%s] ShowTableCreate schema:%s table:%s err:%+v \n", OutputName, schema, table, err)
+	var rows driver.Rows
+	rows, This.err = This.conn.Query(sql, p)
+	if This.err != nil {
+		log.Printf("[ERROR] output[%s] ShowTableCreate schema:%s table:%s err:%+v \n", OutputName, schema, table, This.err)
 		return ""
 	}
 	defer rows.Close()
@@ -231,9 +233,10 @@ func (This *mysqlDB) ShowTableCreate(schema, table string) string {
 func (This *mysqlDB) SelectVersion() string {
 	sql := "SELECT version()"
 	p := make([]driver.Value, 0)
-	rows, err := This.conn.Query(sql, p)
-	if err != nil {
-		log.Printf("[ERROR] output[%s] SelectVersion err:%+v \n", OutputName, err)
+	var rows driver.Rows
+	rows, This.err = This.conn.Query(sql, p)
+	if This.err != nil {
+		log.Printf("[ERROR] output[%s] SelectVersion err:%+v \n", OutputName, This.err)
 		return ""
 	}
 	defer rows.Close()
@@ -253,22 +256,23 @@ func (This *mysqlDB) SelectVersion() string {
 
 func (This *mysqlDB) CreateDatabase(database string) (err error) {
 	sql := fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", database)
-	_, err = This.conn.Exec(sql, []driver.Value{})
-	return
+	_, This.err = This.conn.Exec(sql, []driver.Value{})
+	return This.err
 }
 
 func (This *mysqlDB) Exec(sql string) (err error) {
-	_, err = This.conn.Exec(sql, []driver.Value{})
-	return
+	_, This.err = This.conn.Exec(sql, []driver.Value{})
+	return This.err
 }
 
 func (This *mysqlDB) ShowBackends() (backendsList []map[string]driver.Value, err error) {
 	sql := "SHOW backends"
 	p := make([]driver.Value, 0)
-	rows, err := This.conn.Query(sql, p)
-	if err != nil {
-		log.Printf("[WARN] output[%s] ShowBackbends err:%+v \n", OutputName, err)
-		return make([]map[string]driver.Value, 0), err
+	var rows driver.Rows
+	rows, This.err = This.conn.Query(sql, p)
+	if This.err != nil {
+		log.Printf("[WARN] output[%s] ShowBackbends err:%+v \n", OutputName, This.err)
+		return make([]map[string]driver.Value, 0), This.err
 	}
 	defer rows.Close()
 	for {
@@ -285,3 +289,23 @@ func (This *mysqlDB) ShowBackends() (backendsList []map[string]driver.Value, err
 	}
 	return backendsList, nil
 }
+
+func (This *mysqlDB) ShowVersionComment() (versionComment string, err error) {
+	sql := "SHOW VARIABLES LIKE 'version_comment'"
+	p := make([]driver.Value, 0)
+	rows, err := This.conn.Query(sql, p)
+	if err != nil {
+		log.Printf("[WARN] output[%s] ShowVersionComment err:%+v \n", OutputName, err)
+		return "", err
+	}
+	defer rows.Close()
+	for {
+		dest := make([]driver.Value, len(rows.Columns()), len(rows.Columns()))
+		err := rows.Next(dest)
+		if err != nil {
+			break
+		}
+		versionComment = fmt.Sprint(dest[1])
+	}
+	return
+}

plugin/mysql/src/starrocks.go

@@ -6,6 +6,7 @@ import (
 	pluginDriver "github.com/brokercap/Bifrost/plugin/driver"
 	"log"
 	"runtime/debug"
+	"strings"
 )
 
 func (This *Conn) IsStarRocks() bool {
@@ -17,6 +18,11 @@ func (This *Conn) GetStarRocksBeCount() int {
 }
 
 func (This *Conn) initIsStarrock() {
+	tmpUri := strings.ToLower(*This.uri)
+	if strings.Contains(tmpUri, "starrocks") || strings.Contains(tmpUri, "doris") {
+		This.isStarRocks = true
+		This.starRocksBeCount = 1
+	}
 	defer func() {
 		if err := recover(); err != nil {
 			log.Printf("[ERROR] output[%s] initIsStarrock recover:%+v \n", OutputName, string(debug.Stack()))
@@ -26,16 +32,27 @@ func (This *Conn) initIsStarrock() {
 	if This.conn == nil {
 		return
 	}
-	backendsList, _ := This.conn.ShowBackends()
-	if len(backendsList) == 0 {
-		return
+	if !This.isStarRocks {
+		versionComment, err := This.conn.ShowVersionComment()
+		if err != nil {
+			return
+		}
+		if !strings.Contains(strings.ToLower(versionComment), "mysql") {
+			This.isStarRocks = true
+			This.starRocksBeCount = 1
+		}
 	}
-	// starrocks show backends 列表中,存在 BePort 这个字段,代表 Be 节点的端口
-	if _, ok := backendsList[0]["BePort"]; !ok {
-		return
+	if This.isStarRocks {
+		backendsList, _ := This.conn.ShowBackends()
+		if len(backendsList) == 0 {
+			return
+		}
+		// starrocks show backends 列表中,存在 BePort 这个字段,代表 Be 节点的端口
+		if _, ok := backendsList[0]["BePort"]; !ok {
+			return
+		}
+		This.starRocksBeCount = len(backendsList)
 	}
-	This.isStarRocks = true
-	This.starRocksBeCount = len(backendsList)
 }
 
 func (This *Conn) StarRocksDelete(SchemaName, TableName string, pks []string, pksWhereList [][]string) error {