metrics(cdc): fix mq sink write row count metrics. (#4192) (#4322)

* fix the txn_batch_size metric inaccuracy bug when the sink target is MQ

* address comments

* add comments for exported functions

* fix the compiling problem

* workerpool: limit the rate to output deadlock warning (#3775) (#3799)

* metrics: add changefeed checkepoint catch-up ETA (#3300) (#3311)

* pkg,cdc: do not use log package (#3902) (#3939)

* *: rename repo from pingcap/ticdc to pingcap/tiflow (#3957)

* kvclient(ticdc): fix kvclient takes too long time to recover (#3612) (#3662)

* tz (ticdc): fix timezone error (#3887) (#3910)

* http_*: add log for http api and refine the err handle logic (#2997) (#3306)

* ticdc/alert: add no owner alert rule (#3809) (#3831)

* etcd_worker: batch etcd patch (#3277) (#3393)

* ticdc/owner: Fix ddl special comment syntax error (#3845) (#3977)

* Clean old owner and old processor in release 5.2 branch (#4019)

* tests(ticdc): set up the sync diff output directory correctly (#3725) (#3745)

* owner,scheduler(cdc): fix nil pointer panic in owner scheduler (#2980) (#4007) (#4015)

* config(ticdc): Fix old value configuration check for maxwell protocol (#3747) (#3782)

* ticdc/processor: Fix backoff base delay misconfiguration (#3992) (#4027)

* sink(ticdc): cherry pick sink bug fix to release 5.2 (#4083) (#4119)

* metrics(ticdc): add resolved ts and add changefeed to dataflow (#4038) (#4103)

* kv(ticdc): reduce eventfeed rate limited log (#4072) (#4110)

close #4006

* This is an automated cherry-pick of #4192

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

* http_api (ticdc): fix http api 'get processor' panic. (#4117) (#4122)

close #3840

* cdc/sink: adjust kafka initialization logic (#3192) (#3568)

* This is an automated cherry-pick of #3192

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

* fix conflicts.

* This is an automated cherry-pick of #3682

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

* fix import.

* fix failpoint path.

* try to fix initialize.

* remove table_sink.

* remove initialization.

* remove owner.

* fix mq.

Co-authored-by: Ling Jin <7138436+3AceShowHand@users.noreply.github.com>
Co-authored-by: 3AceShowHand <jinl1037@hotmail.com>

* sink (ticdc): fix a deadlock due to checkpointTs fall back in sinkNode (#4084) (#4098)

close #4055

* This is an automated cherry-pick of #4192

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

* fix conflicts.

* fix conflicts.

* fix conflicts.

Co-authored-by: zhaoxinyu <zhaoxinyu512@gmail.com>
Co-authored-by: amyangfei <yangfei@pingcap.com>
Co-authored-by: dongmen <20351731+asddongmen@users.noreply.github.com>
Co-authored-by: Ling Jin <7138436+3AceShowHand@users.noreply.github.com>
Co-authored-by: 3AceShowHand <jinl1037@hotmail.com>

cdc/sink/codec/avro.go

@@ -120,6 +120,7 @@ func (a *AvroEventBatchEncoder) AppendRowChangedEvent(e *model.RowChangedEvent)
 	}
 
 	mqMessage.Key = evlp
+	mqMessage.IncRowsCount()
 	a.resultBuf = append(a.resultBuf, mqMessage)
 
 	return EncoderNeedAsyncWrite, nil

cdc/sink/codec/canal.go

@@ -377,7 +377,8 @@ func (d *CanalEventBatchEncoder) EncodeDDLEvent(e *model.DDLEvent) (*MQMessage,
 
 // Build implements the EventBatchEncoder interface
 func (d *CanalEventBatchEncoder) Build() []*MQMessage {
-	if len(d.messages.Messages) == 0 {
+	rowCount := len(d.messages.Messages)
+	if rowCount == 0 {
 		return nil
 	}
 
@@ -391,6 +392,7 @@ func (d *CanalEventBatchEncoder) Build() []*MQMessage {
 		log.Panic("Error when serializing Canal packet", zap.Error(err))
 	}
 	ret := NewMQMessage(ProtocolCanal, nil, value, 0, model.MqMessageTypeRow, nil, nil)
+	ret.SetRowsCount(rowCount)
 	d.messages.Reset()
 	d.resetPacket()
 	return []*MQMessage{ret}

cdc/sink/codec/canal_flat.go

@@ -218,7 +218,9 @@ func (c *CanalFlatEventBatchEncoder) Build() []*MQMessage {
 			log.Panic("CanalFlatEventBatchEncoder", zap.Error(err))
 			return nil
 		}
-		ret[i] = NewMQMessage(ProtocolCanalJSON, nil, value, msg.tikvTs, model.MqMessageTypeRow, &msg.Schema, &msg.Table)
+		m := NewMQMessage(ProtocolCanalJSON, nil, value, msg.tikvTs, model.MqMessageTypeRow, &msg.Schema, &msg.Table)
+		m.IncRowsCount()
+		ret[i] = m
 	}
 	c.resolvedBuf = c.resolvedBuf[0:0]
 	return ret

cdc/sink/codec/canal_flat_test.go

@@ -124,6 +124,8 @@ func (s *canalFlatSuite) TestBatching(c *check.C) {
 			c.Assert(msgs, check.HasLen, int(resolvedTs-lastResolved))
 
 			for j := range msgs {
+				c.Assert(msgs[j].GetRowsCount(), check.Equals, 1)
+
 				var msg canalFlatMessage
 				err := json.Unmarshal(msgs[j].Value, &msg)
 				c.Assert(err, check.IsNil)

cdc/sink/codec/canal_test.go

@@ -102,6 +102,7 @@ func (s *canalBatchSuite) TestCanalEventBatchEncoder(c *check.C) {
 		c.Assert(res, check.HasLen, 1)
 		c.Assert(res[0].Key, check.IsNil)
 		c.Assert(len(res[0].Value), check.Equals, size)
+		c.Assert(res[0].GetRowsCount(), check.Equals, len(cs))
 
 		packet := &canal.Packet{}
 		err := proto.Unmarshal(res[0].Value, packet)

cdc/sink/codec/json_test.go

@@ -134,6 +134,7 @@ func (s *batchSuite) testBatchCodec(c *check.C, newEncoder func() EventBatchEnco
 		if len(cs) > 0 {
 			res := encoder.Build()
 			c.Assert(res, check.HasLen, 1)
+			c.Assert(res[0].GetRowsCount(), check.Equals, len(cs))
 			decoder, err := newDecoder(res[0].Key, res[0].Value)
 			c.Assert(err, check.IsNil)
 			checkRowDecoder(decoder, cs)

cdc/sink/codec/maxwell.go

@@ -278,6 +278,7 @@ func (d *MaxwellEventBatchEncoder) Build() []*MQMessage {
 	}
 
 	ret := NewMQMessage(ProtocolMaxwell, d.keyBuf.Bytes(), d.valueBuf.Bytes(), 0, model.MqMessageTypeRow, nil, nil)
+	ret.SetRowsCount(d.batchSize)
 	d.Reset()
 	return []*MQMessage{ret}
 }

cdc/sink/codec/maxwell_test.go

@@ -54,6 +54,7 @@ func (s *maxwellbatchSuite) testmaxwellBatchCodec(c *check.C, newEncoder func()
 			continue
 		}
 		c.Assert(messages, check.HasLen, 1)
+		c.Assert(messages[0].GetRowsCount(), check.Equals, len(cs))
 		c.Assert(len(messages[0].Key)+len(messages[0].Value), check.Equals, size)
 	}