Merge remote-tracking branch 'upstream/main' into kafka-consumer-batch

Signed-off-by: Ping Yu <[email protected]>
tikv · Mar 4, 2024 · 02b5646 · 02b5646
2 parents acf0d83 + 40357a1
commit 02b5646
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/cdc/cdc/sink/mq_test.go b/cdc/cdc/sink/mq_test.go
@@ -198,6 +198,10 @@ func (s mqSinkSuite) TestFlushChangedEvents(c *check.C) {
 	err = sink.EmitChangedEvents(ctx, kv3)
 	c.Assert(err, check.IsNil)
 
+	// TODO: fix EmitCheckpointTs
+	// mock kafka broker processes 1 row resolvedTs event
+	// leader.Returns(prodSuccess)
+
 	checkpointTs1, err := sink.FlushChangedEvents(ctx, keyspanID1, kv1.CRTs)
 	c.Assert(err, check.IsNil)
 	c.Assert(checkpointTs1, check.Equals, kv1.CRTs)

diff --git a/cdc/cdc/sink/producer/kafka/kafka.go b/cdc/cdc/sink/producer/kafka/kafka.go
@@ -260,7 +260,7 @@ func (k *kafkaSaramaProducer) run(ctx context.Context) error {
 			if flushedOffset <= prevOffset {
 				log.Panic("kafka producer flushed offset goes backward", zap.Int32("partition", msg.Partition), zap.Uint64("flushed", flushedOffset), zap.Uint64("prev", prevOffset))
 			}
-			log.Debug("kafka producer flushed message", zap.Int32("partition", msg.Partition), zap.Uint64("offset", flushedOffset))
+			log.Debug("kafka producer flushed message", zap.Int32("partition", msg.Partition), zap.Uint64("offset", flushedOffset), zap.Uint64("prev", prevOffset))
 
 			k.flushedNotifier.Notify()
 		case err := <-k.asyncClient.Errors():