Zichen Liu

Hi @reta Thanks for the contribution! I was wondering if you considered basing your implementation on the [Generic Async Sink](https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink), new in 1.15?