diff --git a/cdc/sinkv2/eventsink/mq/worker.go b/cdc/sinkv2/eventsink/mq/worker.go
index 7edde611ffe..ec183e3f558 100644
--- a/cdc/sinkv2/eventsink/mq/worker.go
+++ b/cdc/sinkv2/eventsink/mq/worker.go
@@ -210,6 +210,7 @@ func (w *worker) batch(
 			return index, nil
 		}
 		if msg.rowEvent != nil {
+			w.statistics.ObserveRows(msg.rowEvent.Event)
 			events[index] = msg
 			index++
 		}
@@ -228,6 +229,7 @@ func (w *worker) batch(
 			}
 
 			if msg.rowEvent != nil {
+				w.statistics.ObserveRows(msg.rowEvent.Event)
 				events[index] = msg
 				index++
 			}