fix canal json sink write row count metrics.

pingcap · Jan 3, 2022 · 519690b · 519690b
1 parent 60ba6a6
commit 519690b
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/cdc/sink/codec/canal_flat.go b/cdc/sink/codec/canal_flat.go
@@ -366,7 +366,9 @@ func (c *CanalFlatEventBatchEncoder) Build() []*MQMessage {
 			log.Panic("CanalFlatEventBatchEncoder", zap.Error(err))
 			return nil
 		}
-		ret[i] = NewMQMessage(config.ProtocolCanalJSON, nil, value, msg.getTikvTs(), model.MqMessageTypeRow, msg.getSchema(), msg.getTable())
+		m := NewMQMessage(config.ProtocolCanalJSON, nil, value, msg.getTikvTs(), model.MqMessageTypeRow, msg.getSchema(), msg.getTable())
+		m.IncRowsCount()
+		ret[i] = m
 	}
 	c.resolvedBuf = c.resolvedBuf[0:0]
 	return ret