[GitHub] [flink] JingsongLi commented on pull request #18412: [FLINK-25696][datastream] Introduce metadataConsumer to InitContext in Sink

[GitBox <gi...@apache.org>]

JingsongLi commented on pull request #18412:
URL: https://github.com/apache/flink/pull/18412#issuecomment-1022113274


   > You can also have consistency problems with the current approach because updating the metadata consumer does not imply that the offsets are committed with the same checkpoint. The commit can either fail and be retried or the `notifyCheckpointComplete` may get lost and the committable is committed with one of the next checkpoints.
   > 
   > Overall I am a bit skeptical about the benefit of the current metadata consumer if the sink works with transactions.
   
   `KakfaProducer.send`: 
   ```
   Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.
   ```
   
   `KakfaProducer.flush`: 
   ```
        * Invoking this method makes all buffered records immediately available to send (even if <code>linger.ms</code> is
        * greater than 0) and blocks on the completion of the requests associated with these records. The post-condition
        * of <code>flush()</code> is that any previously sent record will have completed (e.g. <code>Future.isDone() == true</code>).
        * A request is considered completed when it is successfully acknowledged
        * according to the <code>acks</code> configuration you have specified or else it results in an error.
   ```
   
   And the default value of `acks` is 1, so in this mode, after flushing, the `Callback` must be called. At this point I go to get the offset, there is no consistency problem.
   As for `notifyCheckpointComplete`, the table store will handle the idempotent of commit.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org