add downstream kafka integrate test (#412) (#426)

* tests/* add downstream kafka integrate test

Author: WangXiangUSTC

drainer/config.go

@@ -270,9 +270,6 @@ func (cfg *Config) adjustConfig() error {
 		} else if cfg.SyncerCfg.DestDBType == "pb" {
 			cfg.SyncerCfg.To.BinlogFileDir = cfg.DataDir
 			log.Infof("use default downstream pb directory: %s", cfg.DataDir)
-		} else if cfg.SyncerCfg.DestDBType == "kafka" {
-			cfg.SyncerCfg.To.KafkaAddrs = defaultKafkaAddrs
-			cfg.SyncerCfg.To.KafkaVersion = defaultKafkaVersion
 		}
 	}
 
@@ -298,7 +295,12 @@ func (cfg *Config) adjustConfig() error {
 			cfg.SyncerCfg.To.KafkaVersion = defaultKafkaVersion
 		}
 		if cfg.SyncerCfg.To.KafkaAddrs == "" {
-			cfg.SyncerCfg.To.KafkaAddrs = defaultKafkaAddrs
+			addrs := os.Getenv("KAFKA_ADDRS")
+			if len(addrs) > 0 {
+				cfg.SyncerCfg.To.KafkaAddrs = addrs
+			} else {
+				cfg.SyncerCfg.To.KafkaAddrs = defaultKafkaAddrs
+			}
 		}
 
 		if cfg.SyncerCfg.To.KafkaMaxMessages <= 0 {

go.mod

@@ -62,11 +62,13 @@ require (
 	github.com/petar/GoLLRB v0.0.0-20130427215148-53be0d36a84c // indirect
 	github.com/pierrec/lz4 v2.0.5+incompatible // indirect
 	github.com/pingcap/check v0.0.0-20171206051426-1c287c953996
+	github.com/pingcap/errors v0.11.0 // indirect
 	github.com/pingcap/goleveldb v0.0.0-20161010101021-158edde5a354 // indirect
 	github.com/pingcap/kvproto v0.0.0-20181010074705-0ba3ca8a6e37 // indirect
+	github.com/pingcap/parser v0.0.0-20181210061630-27e9d3e251d4 // indirect
 	github.com/pingcap/pd v2.0.5+incompatible
 	github.com/pingcap/tidb v2.1.0-beta.0.20180823032518-ef6590e1899a+incompatible
-	github.com/pingcap/tidb-tools v0.0.0-20180913093950-ff0ddbf51b57
+	github.com/pingcap/tidb-tools v2.1.1-0.20181130053235-0206fdab9ef8+incompatible
 	github.com/pingcap/tipb v0.0.0-20180711115030-4141907f6909
 	github.com/pkg/errors v0.8.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

tests/README.md

@@ -13,6 +13,7 @@ This folder contains all tests which relies on external service such as TiDB.
 2. The following programs must be installed:
 
    - `mysql`(the CLI client)
+   - `kafka`  working on default port 9092
 
 3. The user executing the tests must have permission to create the folder
 

tests/_utils/run_drainer

@@ -13,7 +13,7 @@ echo "[$(date)] <<<<<< START IN TEST ${TEST_NAME-} FOR: $config >>>>>>" >> "$OUT
 
 if [ -f "$config" ]
 then
-	drainer -config $config -log-file $OUT_DIR/drainer.log >> $OUT_DIR/drainer.log 2>&1
+	drainer -config $config -log-file $OUT_DIR/drainer.log $* >> $OUT_DIR/drainer.log 2>&1
 else
-	drainer -log-file $OUT_DIR/drainer.log >> $OUT_DIR/drainer.log 2>&1
+	drainer -log-file $OUT_DIR/drainer.log $* >> $OUT_DIR/drainer.log 2>&1
 fi

tests/binlog/binlog.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"database/sql"
 	"flag"
 	"os"
 
@@ -24,32 +23,6 @@ func main() {
 		os.Exit(2)
 	}
 
-	TableSQLs := []string{`
-create table ptest(
-	a int primary key, 
-	b double NOT NULL DEFAULT 2.0, 
-	c varchar(10) NOT NULL, 
-	d time unique
-);
-`,
-		`
-create table itest(
-	a int, 
-	b double NOT NULL DEFAULT 2.0, 
-	c varchar(10) NOT NULL, 
-	d time unique,
-	PRIMARY KEY(a, b)
-);
-`,
-		`
-create table ntest(
-	a int, 
-	b double NOT NULL DEFAULT 2.0, 
-	c varchar(10) NOT NULL, 
-	d time unique
-);
-`}
-
 	sourceDB, err := util.CreateDB(cfg.SourceDBCfg)
 	if err != nil {
 		log.Fatal(err)
@@ -62,23 +35,5 @@ create table ntest(
 	}
 	defer util.CloseDB(targetDB)
 
-	// run the simple test case
-	dailytest.RunCase(&cfg.DiffConfig, sourceDB, targetDB)
-
-	dailytest.RunTest(&cfg.DiffConfig, sourceDB, targetDB, func(src *sql.DB) {
-		// generate insert/update/delete sqls and execute
-		dailytest.RunDailyTest(cfg.SourceDBCfg, TableSQLs, cfg.WorkerCount, cfg.JobCount, cfg.Batch)
-	})
-
-	dailytest.RunTest(&cfg.DiffConfig, sourceDB, targetDB, func(src *sql.DB) {
-		// truncate test data
-		dailytest.TruncateTestTable(cfg.SourceDBCfg, TableSQLs)
-	})
-
-	dailytest.RunTest(&cfg.DiffConfig, sourceDB, targetDB, func(src *sql.DB) {
-		// drop test table
-		dailytest.DropTestTable(cfg.SourceDBCfg, TableSQLs)
-	})
-
-	log.Info("test pass!!!")
+	dailytest.Run(sourceDB, targetDB, &cfg.DiffConfig, cfg.WorkerCount, cfg.JobCount, cfg.Batch)
 }

tests/dailytest/dailytest.go

@@ -0,0 +1,58 @@
+package dailytest
+
+import (
+	"database/sql"
+
+	"github.com/ngaut/log"
+	"github.com/pingcap/tidb-binlog/diff"
+)
+
+func Run(sourceDB *sql.DB, targetDB *sql.DB, diffCfg *diff.Config, workerCount int, jobCount int, batch int) {
+
+	TableSQLs := []string{`
+create table ptest(
+	a int primary key,
+	b double NOT NULL DEFAULT 2.0,
+	c varchar(10) NOT NULL,
+	d time unique
+);
+`,
+		`
+create table itest(
+	a int,
+	b double NOT NULL DEFAULT 2.0,
+	c varchar(10) NOT NULL,
+	d time unique,
+	PRIMARY KEY(a, b)
+);
+`,
+		`
+create table ntest(
+	a int,
+	b double NOT NULL DEFAULT 2.0,
+	c varchar(10) NOT NULL,
+	d time unique
+);
+`}
+
+	// run the simple test case
+	RunCase(diffCfg, sourceDB, targetDB)
+
+	RunTest(diffCfg, sourceDB, targetDB, func(src *sql.DB) {
+		// generate insert/update/delete sqls and execute
+		RunDailyTest(sourceDB, TableSQLs, workerCount, jobCount, batch)
+	})
+
+	RunTest(diffCfg, sourceDB, targetDB, func(src *sql.DB) {
+		// truncate test data
+		TruncateTestTable(sourceDB, TableSQLs)
+	})
+
+	RunTest(diffCfg, sourceDB, targetDB, func(src *sql.DB) {
+		// drop test table
+		DropTestTable(sourceDB, TableSQLs)
+	})
+
+	log.Info("test pass!!!")
+
+}

tests/dailytest/exector.go

@@ -1,15 +1,15 @@
 package dailytest
 
 import (
+	"database/sql"
 	"fmt"
 	"sync"
 
 	"github.com/ngaut/log"
-	"github.com/pingcap/tidb-binlog/tests/util"
 )
 
 // RunDailyTest generates insert/update/delete sqls and execute
-func RunDailyTest(dbCfg util.DBConfig, tableSQLs []string, workerCount int, jobCount int, batch int) {
+func RunDailyTest(db *sql.DB, tableSQLs []string, workerCount int, jobCount int, batch int) {
 	var wg sync.WaitGroup
 	wg.Add(len(tableSQLs))
 
@@ -23,32 +23,20 @@ func RunDailyTest(dbCfg util.DBConfig, tableSQLs []string, workerCount int, jobC
 				log.Fatal(err)
 			}
 
-			dbs, err := createDBs(dbCfg, workerCount)
+			err = execSQL(db, tableSQLs[i])
 			if err != nil {
 				log.Fatal(err)
 			}
-			defer closeDBs(dbs)
 
-			err = execSQL(dbs[0], tableSQLs[i])
-			if err != nil {
-				log.Fatal(err)
-			}
-
-			doProcess(table, dbs, jobCount, workerCount, batch)
+			doProcess(table, db, jobCount, workerCount, batch)
 		}(i)
 	}
 
 	wg.Wait()
 }
 
 // TruncateTestTable truncates test data
-func TruncateTestTable(dbCfg util.DBConfig, tableSQLs []string) {
-	db, err := util.CreateDB(dbCfg)
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer util.CloseDB(db)
-
+func TruncateTestTable(db *sql.DB, tableSQLs []string) {
 	for i := range tableSQLs {
 		table := newTable()
 		err := parseTableSQL(table, tableSQLs[i])
@@ -64,13 +52,7 @@ func TruncateTestTable(dbCfg util.DBConfig, tableSQLs []string) {
 }
 
 // DropTestTable drops test table
-func DropTestTable(dbCfg util.DBConfig, tableSQLs []string) {
-	db, err := util.CreateDB(dbCfg)
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer util.CloseDB(db)
-
+func DropTestTable(db *sql.DB, tableSQLs []string) {
 	for i := range tableSQLs {
 		table := newTable()
 		err := parseTableSQL(table, tableSQLs[i])

tests/dailytest/job.go

@@ -96,15 +96,15 @@ func doWait(doneChan chan struct{}, start time.Time, jobCount int, workerCount i
 	close(doneChan)
 }
 
-func doDMLProcess(table *table, dbs []*sql.DB, jobCount int, workerCount int, batch int) {
+func doDMLProcess(table *table, db *sql.DB, jobCount int, workerCount int, batch int) {
 	jobChan := make(chan struct{}, 16*workerCount)
 	doneChan := make(chan struct{}, workerCount)
 
 	start := time.Now()
 	go addJobs(jobCount, jobChan)
 
 	for i := 0; i < workerCount; i++ {
-		go doJob(table, dbs[i], batch, jobChan, doneChan)
+		go doJob(table, db, batch, jobChan, doneChan)
 	}
 
 	doWait(doneChan, start, jobCount, workerCount)
@@ -149,12 +149,12 @@ func doDDLProcess(table *table, db *sql.DB) {
 	execSqls(db, []string{sql}, [][]interface{}{{}})
 }
 
-func doProcess(table *table, dbs []*sql.DB, jobCount int, workerCount int, batch int) {
+func doProcess(table *table, db *sql.DB, jobCount int, workerCount int, batch int) {
 	if len(table.columns) <= 2 {
 		log.Fatal("column count must > 2, and the first and second column are for primary key")
 	}
 
-	doDMLProcess(table, dbs, jobCount/2, workerCount, batch)
-	doDDLProcess(table, dbs[0])
-	doDMLProcess(table, dbs, jobCount/2, workerCount, batch)
+	doDMLProcess(table, db, jobCount/2, workerCount, batch)
+	doDDLProcess(table, db)
+	doDMLProcess(table, db, jobCount/2, workerCount, batch)
 }

tests/flash/flash.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"database/sql"
 	"flag"
 	"fmt"
 	"os"
@@ -27,32 +26,6 @@ func main() {
 		os.Exit(2)
 	}
 
-	TableSQLs := []string{`
-create table ptest(
-	a int primary key, 
-	b double NOT NULL DEFAULT 2.0, 
-	c varchar(10) NOT NULL, 
-	d time unique
-);
-`,
-		`
-create table itest(
-	a int, 
-	b double NOT NULL DEFAULT 2.0, 
-	c varchar(10) NOT NULL, 
-	d time unique,
-	PRIMARY KEY(a, b)
-);
-`,
-		`
-create table ntest(
-	a int, 
-	b double NOT NULL DEFAULT 2.0, 
-	c varchar(10) NOT NULL, 
-	d time unique
-);
-`}
-
 	sourceDB, err := util.CreateDB(cfg.SourceDBCfg)
 	if err != nil {
 		log.Fatal(err)
@@ -84,23 +57,5 @@ create table ntest(
 		log.Fatal(err)
 	}
 
-	// run the simple test case
-	dailytest.RunCase(&cfg.DiffConfig, sourceDB, targetDB)
-
-	dailytest.RunTest(&cfg.DiffConfig, sourceDB, targetDB, func(src *sql.DB) {
-		// generate insert/update/delete sqls and execute
-		dailytest.RunDailyTest(cfg.SourceDBCfg, TableSQLs, cfg.WorkerCount, cfg.JobCount, cfg.Batch)
-	})
-
-	dailytest.RunTest(&cfg.DiffConfig, sourceDB, targetDB, func(src *sql.DB) {
-		// truncate test data
-		dailytest.TruncateTestTable(cfg.SourceDBCfg, TableSQLs)
-	})
-
-	dailytest.RunTest(&cfg.DiffConfig, sourceDB, targetDB, func(src *sql.DB) {
-		// drop test table
-		dailytest.DropTestTable(cfg.SourceDBCfg, TableSQLs)
-	})
-
-	log.Info("test pass!!!")
+	dailytest.Run(sourceDB, targetDB, &cfg.DiffConfig, cfg.WorkerCount, cfg.JobCount, cfg.Batch)
 }

tests/kafka/drainer.toml

@@ -0,0 +1,35 @@
+# drainer meta data directory path
+data-dir = "/tmp/tidb_binlog_test/data.drainer"
+
+# a comma separated list of PD endpoints
+pd-urls = "http://127.0.0.1:2379"
+
+# syncer Configuration.
+[syncer]
+
+# disable sync these schema
+ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
+
+# number of binlog events in a transaction batch
+txn-batch = 1
+
+# work count to execute binlogs
+worker-count = 1
+
+disable-dispatch = false
+
+# safe mode will split update to delete and insert
+safe-mode = false
+
+# downstream storage, equal to --dest-db-type
+# valid values are "mysql", "pb", "tidb", "flash", "kafka"
+db-type = "kafka"
+
+# Time and size limits for flash batch write
+# time-limit = "30s"
+# size-limit = "100000"
+[syncer.to.checkpoint]
+#schema = "tidb_binlog"
+
+[syncer.to]
+topic-name = "binlog_test_topic"