mounter: fix the block in WaitPrepare function (pingcap#449)

5kbpers · Apr 10, 2020 · f7f9435 · f7f9435
1 parent 6ef44cc
commit f7f9435
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 4 deletions.
diff --git a/cdc/model/mounter.go b/cdc/model/mounter.go
@@ -13,6 +13,8 @@
 
 package model
 
+import "context"
+
 // PolymorphicEvent describes a event can be in multiple states
 type PolymorphicEvent struct {
 	Ts       uint64
@@ -52,8 +54,13 @@ func (e *PolymorphicEvent) PrepareFinished() {
 }
 
 // WaitPrepare waits for prepare process finished
-func (e *PolymorphicEvent) WaitPrepare() {
+func (e *PolymorphicEvent) WaitPrepare(ctx context.Context) error {
 	if e.finished != nil {
-		<-e.finished
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-e.finished:
+		}
 	}
+	return nil
 }
diff --git a/cdc/processor.go b/cdc/processor.go
@@ -550,11 +550,14 @@ func (p *processor) syncResolved(ctx context.Context) error {
 	for {
 		select {
 		case row := <-p.output:
-			row.WaitPrepare()
+			err := row.WaitPrepare(ctx)
+			if err != nil {
+				return errors.Trace(err)
+			}
 			if row.Row == nil {
 				continue
 			}
-			err := p.sink.EmitRowChangedEvent(ctx, row.Row)
+			err = p.sink.EmitRowChangedEvent(ctx, row.Row)
 			if err != nil {
 				return errors.Trace(err)
 			}