fix maxwell.

pingcap · Jan 4, 2022 · 56df902 · 56df902
1 parent 543bd4d
commit 56df902
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 0 deletions.
diff --git a/cdc/sink/codec/maxwell.go b/cdc/sink/codec/maxwell.go
@@ -298,6 +298,7 @@ func (d *MaxwellEventBatchEncoder) Build() []*MQMessage {
 	}
 
 	ret := NewMQMessage(config.ProtocolMaxwell, d.keyBuf.Bytes(), d.valueBuf.Bytes(), 0, model.MqMessageTypeRow, nil, nil)
+	ret.SetRowsCount(d.batchSize)
 	d.Reset()
 	return []*MQMessage{ret}
 }

diff --git a/cdc/sink/codec/maxwell_test.go b/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)
 	}