Enhance mqSink to support Low-latency/DDL-consistent/Txn-consistent modes

[liuzix]

Feature Request

Is your feature request related to a problem? Please describe:

• Canal support is not DDL-consistent in the sense that a DDL might be reordered with DMLs.
• In order to make Canal support DDL-consistent, we sacrifice latency because we need to wait until FlushRowChangedEvents is issued.
• If we write transaction reconstruction logic in mqSink, code complexity will be increased, and because different encoders might want to handle transactions differently, letting mqSink construct the transactions hampers the decoupling of code.

Describe the feature you'd like:

• DDL-consistent mode should be achieved by preventing the mqProducer from implicit flushes, i.e. write to the MQ without explicit Flush calls. Hence I propose writing a mqProducer decorator that can make any producer hold back sending data to network until Flush is called.
• Txn-consistent mode should be supported directly by encoders, which should reconstruct the transactions internally. To this end, we should provide TxnGenerator that facilitates the reconstruction of transactions. TxnGenerator should support Append and Split methods. Changes in Refactor: Let the encoder control sink's output related operation #770 should be merged so that the encoder can control when it wants the sink to write to the MQ.

Current support/needs

	Open Protocol	Canal	Avro
Low-latency	Available	Available	Available
DDL-consistent	No plan	Needed to ensure correctness	No need because there's no DDL output
Transaction-consistent	Handled by consumer	Planned	No need because there's no semantics

[amyangfei]

Have one problem with the DDL operation in canal, if we dispatch DML data to multiple partitions, and DDL to the first partition, how does the consumer know when to consume DDL from the first partition. From another aspect, how does canal consumer know which DMLs are before one DDL, and which DMLs are after one DDL.
Btw, I found a similar scenario about DML, DDL consumption with multiple MQ partitions. alibaba/canal#2842

[zier-one]

also, see https://github.com/pingcap/ticdc/pull/708/files#diff-13ef8e2f0f85e13cd6f76c33fa8d982dR28,

there is a simple abstraction and implement about the transaction-consistent controller