[3.0][cdc-runtime] add DataSinkWriterOperator to process Event

[lvyanquan]

close #2605

points to be discussed:

• should we add a new interface SupportSchemaEvolutionWriting, or let writer deal with it in write method？
• Unnecessary.
• is there a better way to access SinkWriter rather than reflection, as sinkWriter is private scope？
• Reflection is acceptable.
• SCHEMA_EVOLUTION_OPERATOR_ID is an identification to get SchemaOperatorCoordinator, does it meet the expectations of SchemaOperator？
• OperatorID of SchemaOperator is created by Composer.

@PatrickRen @ruanhang1993 cc and Looking forward to your suggestions.

[PatrickRen]

@lvyanquan Thanks for the pull request!

1. I don't quite get the usage of the interface SupportSchemaEvolutionWriting. Which component should this interface be applied to?
2. Using reflections is acceptable to me. I couldn't find better solution for now. Otherwise we have to copy-paste and rewrite the entire SinkWriterOperator from Flink.
3. I'm not sure if it is possible to build the operator ID in composing phase, instead of hard-coding in schema operator coordinator. It looks like the composer's responsibility to generate a ID and use it to construct the schema operator and the sink operator.

[lvyanquan]

@PatrickRen

1. I don't quite get the usage of the interface SupportSchemaEvolutionWriting. Which component should this interface be applied to?

SupportSchemaEvolutionWriting is designed for SinkWriter, just to separate the processing logic of SchemaChangeEvent and
DataChangeEvent. But it's somewhat redundant, and was deleted now.

3. I'm not sure if it is possible to build the operator ID in composing phase, instead of hard-coding in schema operator coordinator. It looks like the composer's responsibility to generate a ID and use it to construct the schema operator and the sink operator.

Keep this code currently until another implementation is available

[ruanhang1993]

LGTM.

[PatrickRen]

@lvyanquan Thanks for the pull request! I left some comments.

[PatrickRen]

Two tiny comments for the operator

[PatrickRen]

@lvyanquan Sorry for flushing your thread! Found another issue.

[PatrickRen]

@lvyanquan Thanks for the update! LGTM.