refactor drainer syncer

Makefile

@@ -99,12 +99,8 @@ else
 	grep -F '<option' "$(TEST_DIR)/all_cov.html"
 endif
 
-
 check-static:
-	gometalinter --disable-all --deadline 120s \
-		--enable megacheck \
-		--enable ineffassign \
-		$$($(PACKAGE_DIRECTORIES))
+	golangci-lint --disable errcheck run $$($(PACKAGE_DIRECTORIES))
 
 update: update_vendor clean_vendor
 update_vendor:

drainer/checkpoint/checkpoint.go

@@ -55,7 +55,7 @@ func NewCheckPoint(name string, cfg *Config) (CheckPoint, error) {
 	case "mysql", "tidb":
 		cp, err = newMysql(name, cfg)
 	case "pb":
-		cp, err = newPb(cfg)
+		cp, err = NewPb(cfg)
 	case "kafka":
 		cp, err = newKafka(cfg)
 	case "flash":

drainer/checkpoint/kafka.go

@@ -14,7 +14,6 @@
 package checkpoint
 
 import (
-	"math"
 	"sync"
 
 	"github.com/ngaut/log"
@@ -25,56 +24,18 @@ import (
 type KafkaCheckpoint struct {
 	sync.Mutex
 	*PbCheckPoint
-
-	meta *KafkaMeta
-}
-
-// KafkaMeta maintains the safeTS which is synced to kafka
-type KafkaMeta struct {
-	// set to be the minimize commit ts of binlog if we have some binlog not having successful sent to kafka
-	// or set to be `math.MaxInt64`
-	safeTS int64
-
-	lock sync.Mutex
-	cond *sync.Cond
-}
-
-// SetSafeTS set the safe ts to be ts
-func (m *KafkaMeta) SetSafeTS(ts int64) {
-	log.Debug("set safe ts: ", ts)
-
-	m.lock.Lock()
-	m.safeTS = ts
-	m.cond.Signal()
-	m.lock.Unlock()
-}
-
-var kafkaMeta *KafkaMeta
-var kafkaMetaOnce sync.Once
-
-// GetKafkaMeta return the singleton instance of KafkaMeta
-func GetKafkaMeta() *KafkaMeta {
-	kafkaMetaOnce.Do(func() {
-		kafkaMeta = new(KafkaMeta)
-		kafkaMeta.cond = sync.NewCond(&kafkaMeta.lock)
-	})
-
-	return kafkaMeta
 }
 
 func newKafka(cfg *Config) (CheckPoint, error) {
-	pb, err := newPb(cfg)
+	pb, err := NewPb(cfg)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
 	cp := &KafkaCheckpoint{
 		PbCheckPoint: pb.(*PbCheckPoint),
-		meta:         GetKafkaMeta(),
 	}
 
-	cp.meta.SetSafeTS(math.MaxInt64)
-
 	return cp, nil
 }
 
@@ -92,12 +53,5 @@ func (cp *KafkaCheckpoint) Save(ts int64) error {
 		return nil
 	}
 
-	cp.meta.lock.Lock()
-	for cp.meta.safeTS < ts {
-		log.Debugf("wait for ts: %d safe_ts: %d", ts, cp.meta.safeTS)
-		cp.meta.cond.Wait()
-	}
-	cp.meta.lock.Unlock()
-
 	return cp.PbCheckPoint.Save(ts)
 }

drainer/checkpoint/kafka_test.go

@@ -15,7 +15,6 @@ package checkpoint
 
 import (
 	"path/filepath"
-	"time"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
@@ -36,25 +35,6 @@ func (t *testCheckPointSuite) TestKafka(c *C) {
 	ts := cp.TS()
 	c.Assert(ts, Equals, testTs)
 
-	// test for safeTs
-	cp2, ok := cp.(*KafkaCheckpoint)
-	c.Assert(ok, IsTrue)
-
-	var safeTs int64 = 100
-	newTs := safeTs + 1
-	cp2.meta.SetSafeTS(safeTs)
-
-	go func() {
-		time.Sleep(500 * time.Millisecond) // sleep for a while
-		cp2.meta.SetSafeTS(newTs)
-	}()
-
-	begin := time.Now()
-	err = cp.Save(newTs) // block until `newTs` be set
-	c.Assert(err, IsNil)
-	c.Assert(cp.TS(), Equals, newTs)
-	c.Assert(time.Since(begin).Seconds(), Greater, 0.49) // ~ 0.5
-
 	// close the checkpoint
 	err = cp.Close()
 	c.Assert(err, IsNil)

drainer/checkpoint/pb.go

@@ -43,7 +43,7 @@ type PbCheckPoint struct {
 }
 
 // NewPb creates a new Pb.
-func newPb(cfg *Config) (CheckPoint, error) {
+func NewPb(cfg *Config) (CheckPoint, error) {
 	pb := &PbCheckPoint{initialCommitTS: cfg.InitialCommitTS, name: cfg.CheckPointFile, saveTime: time.Now()}
 	err := pb.Load()
 	if err != nil {

drainer/checkpoint/pb_test.go

@@ -26,7 +26,7 @@ func (t *testCheckPointSuite) TestPb(c *C) {
 	notExistFileName := "test_not_exist"
 	cfg := new(Config)
 	cfg.CheckPointFile = fileName
-	meta, err := newPb(cfg)
+	meta, err := NewPb(cfg)
 	c.Assert(err, IsNil)
 	defer os.RemoveAll(fileName)
 
@@ -49,15 +49,15 @@ func (t *testCheckPointSuite) TestPb(c *C) {
 
 	// check not exist meta file
 	cfg.CheckPointFile = notExistFileName
-	meta, err = newPb(cfg)
+	meta, err = NewPb(cfg)
 	c.Assert(err, IsNil)
 	err = meta.Load()
 	c.Assert(err, IsNil)
 	c.Assert(meta.TS(), Equals, int64(0))
 
 	// check not exist meta file, but with initialCommitTs
 	cfg.InitialCommitTS = 123
-	meta, err = newPb(cfg)
+	meta, err = NewPb(cfg)
 	c.Assert(err, IsNil)
 	c.Assert(meta.TS(), Equals, cfg.InitialCommitTS)
 

drainer/config.go

@@ -30,7 +30,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/mysql"
 
-	"github.com/pingcap/tidb-binlog/drainer/executor"
+	dsync "github.com/pingcap/tidb-binlog/drainer/sync"
 	"github.com/pingcap/tidb-binlog/pkg/filter"
 	"github.com/pingcap/tidb-binlog/pkg/flags"
 	"github.com/pingcap/tidb-binlog/pkg/security"
@@ -64,7 +64,7 @@ type SyncerConfig struct {
 	IgnoreTables     []filter.TableName `toml:"ignore-table" json:"ignore-table"`
 	TxnBatch         int                `toml:"txn-batch" json:"txn-batch"`
 	WorkerCount      int                `toml:"worker-count" json:"worker-count"`
-	To               *executor.DBConfig `toml:"to" json:"to"`
+	To               *dsync.DBConfig    `toml:"to" json:"to"`
 	DoTables         []filter.TableName `toml:"replicate-do-table" json:"replicate-do-table"`
 	DoDBs            []string           `toml:"replicate-do-db" json:"replicate-do-db"`
 	DestDBType       string             `toml:"db-type" json:"db-type"`
@@ -282,7 +282,7 @@ func (cfg *Config) adjustConfig() error {
 
 	// add default syncer.to configuration if need
 	if cfg.SyncerCfg.To == nil {
-		cfg.SyncerCfg.To = new(executor.DBConfig)
+		cfg.SyncerCfg.To = new(dsync.DBConfig)
 	}
 	if cfg.SyncerCfg.DestDBType == "kafka" {
 		// get KafkaAddrs from zookeeper if ZkAddrs is setted