Add a channel-based interface for sink to allow batch processing

cdc/sink/mysql.go

@@ -38,6 +38,12 @@ import (
 	"github.com/pingcap/tidb/types"
 )
 
+// TableInfoGetter is used to get table info by table id of TiDB
+type TableInfoGetter interface {
+	TableByID(id int64) (info *model.TableInfo, ok bool)
+	GetTableIDByName(schema, table string) (int64, bool)
+}
+
 type tableInspector interface {
 	// Get returns information about the specified table
 	Get(schema, table string) (*tableInfo, error)
@@ -47,6 +53,8 @@ type tableInspector interface {
 
 type mysqlSink struct {
 	db           *sql.DB
+	successC     chan txn.Txn
+	errC         chan error
 	tblInspector tableInspector
 	infoGetter   TableInfoGetter
 }
@@ -67,6 +75,8 @@ func NewMySQLSink(
 	sink := mysqlSink{
 		db:           db,
 		infoGetter:   infoGetter,
+		successC:     make(chan txn.Txn, 1),
+		errC:         make(chan error, 1),
 		tblInspector: cachedInspector,
 	}
 	return &sink, nil
@@ -83,11 +93,49 @@ func NewMySQLSinkUsingSchema(db *sql.DB, picker *schema.Schema) Sink {
 	}
 	return &mysqlSink{
 		db:           db,
+		successC:     make(chan txn.Txn, 1),
+		errC:         make(chan error, 1),
 		infoGetter:   picker,
 		tblInspector: inspector,
 	}
 }
 
+func (s *mysqlSink) Run(ctx context.Context, input <-chan txn.Txn) error {
+	for {
+		select {
+		case t, ok := <-input:
+			if !ok {
+				log.Info("Input channel closed")
+				return nil
+			}
+			if err := s.Emit(ctx, t); err != nil {
+				return err
+			}
+			select {
+			case s.successC <- t:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
+
+func (s *mysqlSink) Success() <-chan txn.Txn {
+	return s.successC
+}
+
+func (s *mysqlSink) Error() <-chan error {
+	return s.errC
+}
+
+func (s *mysqlSink) Close() error {
+	close(s.successC)
+	close(s.errC)
+	return s.db.Close()
+}
+
 func (s *mysqlSink) Emit(ctx context.Context, txn txn.Txn) error {
 	filterBySchemaAndTable(&txn)
 	if len(txn.DMLs) == 0 && txn.DDL == nil {
@@ -126,18 +174,6 @@ func filterBySchemaAndTable(t *txn.Txn) {
 	}
 }
 
-func (s *mysqlSink) EmitResolvedTimestamp(ctx context.Context, resolved uint64) error {
-	return nil
-}
-
-func (s *mysqlSink) Flush(ctx context.Context) error {
-	return nil
-}
-
-func (s *mysqlSink) Close() error {
-	return nil
-}
-
 func (s *mysqlSink) execDDLWithMaxRetries(ctx context.Context, ddl *txn.DDL, maxRetries uint64) error {
 	retryCfg := backoff.WithMaxRetries(
 		backoff.WithContext(

cdc/sink/mysql_test.go

@@ -15,6 +15,7 @@ package sink
 
 import (
 	"context"
+	"time"
 
 	dmysql "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/parser/mysql"
@@ -106,6 +107,59 @@ func (s EmitSuite) TestShouldIgnoreCertainDDLError(c *check.C) {
 	c.Assert(mock.ExpectationsWereMet(), check.IsNil)
 }
 
+type RunSuite struct{}
+
+var _ = check.Suite(&RunSuite{})
+
+func (rs *RunSuite) TestShouldSendSuccessTxns(c *check.C) {
+	// Set up
+	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
+	c.Assert(err, check.IsNil)
+	defer db.Close()
+
+	sink := mysqlSink{
+		db:           db,
+		tblInspector: dummyInspector{},
+		successC:     make(chan txn.Txn, 1),
+		errC:         make(chan error, 1),
+	}
+
+	t := txn.Txn{
+		DDL: &txn.DDL{
+			Database: "test",
+			Table:    "user",
+			SQL:      "CREATE TABLE user (id INT PRIMARY KEY);",
+		},
+	}
+
+	mock.ExpectBegin()
+	mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectExec(t.DDL.SQL).WillReturnResult(sqlmock.NewResult(1, 1))
+	mock.ExpectCommit()
+
+	ctx := context.Background()
+	input := make(chan txn.Txn, 1)
+	input <- t
+	close(input)
+	stopped := make(chan struct{})
+	go func() {
+		err = sink.Run(ctx, input)
+		c.Assert(err, check.IsNil)
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(time.Second):
+		c.Fatal("Run didn't stop in time")
+	}
+	select {
+	case txn := <-sink.Success():
+		c.Assert(t, check.DeepEquals, txn)
+	default:
+		c.Fatal("Success txn not sent")
+	}
+}
+
 type tableHelper struct {
 	tableInspector
 	TableInfoGetter

cdc/sink/sink.go

@@ -15,54 +15,16 @@ package sink
 
 import (
 	"context"
-	"fmt"
-	"io"
 
-	"github.com/pingcap/parser/model"
 	"github.com/pingcap/tidb-cdc/cdc/txn"
 )
 
 // Sink is an abstraction for anything that a changefeed may emit into.
 type Sink interface {
-	Emit(ctx context.Context, t txn.Txn) error
-	EmitResolvedTimestamp(
-		ctx context.Context,
-		resolved uint64,
-	) error
-	// TODO: Add GetLastSuccessTs() uint64
-	// Flush blocks until every message enqueued by EmitRow and
-	// EmitResolvedTimestamp has been acknowledged by the sink.
-	Flush(ctx context.Context) error
-	// Close does not guarantee delivery of outstanding messages.
+	Run(ctx context.Context, txns <-chan txn.Txn) error
+	Success() <-chan txn.Txn
+	Error() <-chan error
 	Close() error
-}
-
-// TableInfoGetter is used to get table info by table id of TiDB
-type TableInfoGetter interface {
-	TableByID(id int64) (info *model.TableInfo, ok bool)
-	GetTableIDByName(schema, table string) (int64, bool)
-}
-
-type writerSink struct {
-	io.Writer
-}
-
-var _ Sink = &writerSink{}
-
-func (s *writerSink) Emit(ctx context.Context, t txn.Txn) error {
-	fmt.Fprintf(s, "commit ts: %d", t.Ts)
-	return nil
-}
-
-func (s *writerSink) EmitResolvedTimestamp(ctx context.Context, resolved uint64) error {
-	fmt.Fprintf(s, "resolved: %d", resolved)
-	return nil
-}
-
-func (s *writerSink) Flush(ctx context.Context) error {
-	return nil
-}
-
-func (s *writerSink) Close() error {
-	return nil
+	// TODO: Replace Emit completely with Run
+	Emit(ctx context.Context, txn txn.Txn) error
 }