sink(ticdc): fix TxnEventAppender.startTs in CreateTableSink (pingcap…

…#8810)

close pingcap#8805

cdc/sinkv2/eventsink/factory/factory.go

@@ -91,11 +91,8 @@ func New(ctx context.Context,
 	return s, nil
 }
 
-// CreateTableSinkForConsumer creates a TableSink by schema for consumer.
-// The difference between CreateTableSink and CreateTableSinkForConsumer is that
-// CreateTableSinkForConsumer will not create a new sink for each table.
-// NOTICE: This only used for the consumer. Please do not use it in the processor.
-func (s *SinkFactory) CreateTableSinkForConsumer(
+// CreateTableSink creates a TableSink by schema.
+func (s *SinkFactory) CreateTableSink(
 	changefeedID model.ChangeFeedID,
 	tableID model.TableID, startTs model.Ts,
 	totalRowsCounter prometheus.Counter,
@@ -106,18 +103,18 @@ func (s *SinkFactory) CreateTableSinkForConsumer(
 		return tablesink.New[*model.RowChangedEvent](changefeedID, tableID, startTs,
 			s.rowSink, &eventsink.RowChangeEventAppender{}, totalRowsCounter)
 	case sink.TxnSink:
-		return tablesink.New[*model.SingleTableTxn](changefeedID, tableID, startTs, s.txnSink,
-			// IgnoreStartTs is true because the consumer can
-			// **not** get the start ts of the row changed event.
-			&eventsink.TxnEventAppender{TableSinkStartTs: startTs, IgnoreStartTs: true},
-			totalRowsCounter)
+		return tablesink.New[*model.SingleTableTxn](changefeedID, tableID, startTs,
+			s.txnSink, &eventsink.TxnEventAppender{TableSinkStartTs: startTs}, totalRowsCounter)
 	default:
 		panic("unknown sink type")
 	}
 }
 
-// CreateTableSink creates a TableSink by schema.
-func (s *SinkFactory) CreateTableSink(
+// CreateTableSinkForConsumer creates a TableSink by schema for consumer.
+// The difference between CreateTableSink and CreateTableSinkForConsumer is that
+// CreateTableSinkForConsumer will not create a new sink for each table.
+// NOTICE: This only used for the consumer. Please do not use it in the processor.
+func (s *SinkFactory) CreateTableSinkForConsumer(
 	changefeedID model.ChangeFeedID,
 	tableID model.TableID, startTs model.Ts,
 	totalRowsCounter prometheus.Counter,
@@ -128,8 +125,11 @@ func (s *SinkFactory) CreateTableSink(
 		return tablesink.New[*model.RowChangedEvent](changefeedID, tableID, startTs,
 			s.rowSink, &eventsink.RowChangeEventAppender{}, totalRowsCounter)
 	case sink.TxnSink:
-		return tablesink.New[*model.SingleTableTxn](changefeedID, tableID, startTs,
-			s.txnSink, &eventsink.TxnEventAppender{}, totalRowsCounter)
+		return tablesink.New[*model.SingleTableTxn](changefeedID, tableID, startTs, s.txnSink,
+			// IgnoreStartTs is true because the consumer can
+			// **not** get the start ts of the row changed event.
+			&eventsink.TxnEventAppender{TableSinkStartTs: startTs, IgnoreStartTs: true},
+			totalRowsCounter)
 	default:
 		panic("unknown sink type")
 	}