fix(sink(ticdc)): fix dispatch error

pingcap · Jan 16, 2022 · ab62e6b · ab62e6b
1 parent 36e4685
commit ab62e6b
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/cdc/sink/mysql.go b/cdc/sink/mysql.go
@@ -434,9 +434,10 @@ func (s *mysqlSink) broadcastFinishTxn() {
 }
 
 func (s *mysqlSink) dispatchTableTxn(ctx context.Context, txn *model.SingleTableTxn) {
+	log.Info("dispatch single table txn", zap.Uint64("startTs", txn.RawTableTxn.StartTs), zap.Uint64("commitTs", txn.RawTableTxn.CommitTs))
 	// NOTICE: To some extend, causality dispatcher may affect other dispatcher due to the waitExec
 	workerIdx := s.txnDispatcher.Dispatch(&(txn.RawTableTxn))
-	if workerIdx > 0 {
+	if workerIdx >= 0 {
 		s.workers[workerIdx].appendTxn(ctx, txn)
 		return
 	}