drainer: large binlog may skipped when encountered "Message was too large" error

[sleepymole]

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do?
   a. Use drainer to sync binlog to kafka.
   b. Insert a large amount of data in a transaction.

2. What did you expect to see?
   "Message was too large, server rejected it to avoid allocation error" occurred in drainer no matter how many times the drainer restarts.

3. What did you see instead?
   After several restarts, drainer didn't report any error. And save point was increased normally.
   The root cause may be that the error item cannot be received from p.producer.Successes(). If there are small binlog after the large binlog, it can be write to kafka normally and return success item to syncer.
   

   tidb-binlog/drainer/sync/kafka.go

   Lines 195 to 200 in b607f34

   	for msg := range p.producer.Successes() {
   	item := msg.Metadata.(*Item)
   	commitTs := item.Binlog.GetCommitTs()
   	log.Debug("get success msg from producer", zap.Int64("ts", commitTs))
   	p.toBeAckCommitTSMu.Lock()

4. Please provide the relate downstream type and version of drainer.
   (run drainer -V in terminal to get drainer's version)
   v4.0.6 (this issue may exist in all versions)