sink: wait prepare before using Row

Signed-off-by: Neil Shen <overvenus@gmail.com>
pingcap · Aug 26, 2021 · bbfda29 · bbfda29
1 parent ba5b0eb
commit bbfda29
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/cdc/processor/pipeline/sink.go b/cdc/processor/pipeline/sink.go
@@ -149,7 +149,14 @@ func (n *sinkNode) flushSink(ctx pipeline.NodeContext, resolvedTs model.Ts) (err
 }
 
 func (n *sinkNode) emitEvent(ctx pipeline.NodeContext, event *model.PolymorphicEvent) error {
-	if event == nil || event.Row == nil {
+	if event == nil {
+		log.Warn("skip emit empty rows", zap.Reflect("event", event))
+		return nil
+	}
+	if err := event.WaitPrepare(ctx); err != nil {
+		return err
+	}
+	if event.Row == nil {
 		log.Warn("skip emit empty rows", zap.Reflect("event", event))
 		return nil
 	}