simple (ticdc): fix bootstrap event is not sent (#11031)

close #11032
pingcap · May 9, 2024 · 91a3cb2 · 91a3cb2
1 parent eeb9517
commit 91a3cb2
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/cdc/sink/dmlsink/mq/mq_dml_sink.go b/cdc/sink/dmlsink/mq/mq_dml_sink.go
@@ -190,9 +190,10 @@ func (s *dmlSink) WriteEvents(txns ...*dmlsink.CallbackableEvent[*model.SingleTa
 			// This never be blocked because this is an unbounded channel.
 			s.alive.worker.msgChan.In() <- mqEvent{
 				key: model.TopicPartitionKey{
-					Topic:        topic,
-					Partition:    index,
-					PartitionKey: key,
+					Topic:          topic,
+					Partition:      index,
+					PartitionKey:   key,
+					TotalPartition: partitionNum,
 				},
 				rowEvent: &dmlsink.RowChangeCallbackableEvent{
 					Event:     row,

diff --git a/pkg/sink/codec/bootstraper.go b/pkg/sink/codec/bootstraper.go
@@ -75,6 +75,7 @@ func newBootstrapWorker(
 }
 
 func (b *bootstrapWorker) run(ctx context.Context) error {
+	log.Info("Bootstrap worker is started", zap.Stringer("changefeed", b.changefeedID))
 	sendTicker := time.NewTicker(bootstrapWorkerTickerInterval)
 	gcTicker := time.NewTicker(bootstrapWorkerGCInterval)
 	defer func() {