[Flink] Make flink sink generic

[polyzos]

This PR makes the Flink Sink generic so it can later be used by the FlussSink for the datastream API and handle various data types. It also introduces a ConvertingSinkWriter for converting between various types and RowData.

[wuchong]

Thanks, @polyzos. I’m wondering why we aren’t introducing a FlussSerializationSchema that directly converts T into InternalRow. This approach would be consistent with how we handled the generic source, where we directly worked with InternalRow.

In this PR, we introduced the ConvertingSinkWriter, which first converts T into RowData and then subsequently converts RowData into InternalRow. However, the intermediate conversion to RowData seems unnecessary since our ultimate goal is to work with InternalRow directly.

Wouldn’t it make more sense to bypass RowData entirely and streamline the process by converting T directly into InternalRow?

[polyzos]

@wuchong ConvertingSinkWriter was introduced for convenience, as I was getting some compilation issues with the current implementation. I figured out indeed a better approach though... since this PR is mainly around introducing a ConvertingSinkWriter to achieve make the class generic, I will close this and open a new PR from latest branch to avoid deletion etc.