[FLINK-11320] Support user callback in Kafka sink

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Component/s: Connectors / Kafka
• Labels: auto-deprioritized-major auto-deprioritized-minor pull-request-available
• Assignee: Unassigned
• Reporter: Allen Wang
• Created: 14/Jan/19 21:49
• Updated: 30/Nov/21 10:42

Description

We would like to get notified when Flink's Kafka sink successfully produces a message to the Kafka cluster for message tracing purposes. The callback should be created for the specific item being produced. Currently this functionality is inside Flink's internal code and not accessible to the user. 

This change would allow user to get callback when a message is produced in the sink:

• Add a new interface UserCallbackFactory:

  public interface UserCallbackFactory<IN> {
      Callback create(IN next);
  }

• Add new constructor in FlinkKafkaProducer to pass in an optional argument of UserCallbackFactory
• In the  FlinkKafkaProducer.invoke method, create a user callback given the produced item and invoke this callback.

 

 

Issue Links

• links to: GitHub Pull Request #7877