[GitHub] [flink] tzulitai commented on a diff in pull request #22150: [FLINK-31305] Propagate producer exceptions outside of mailbox execut…

["tzulitai (via GitHub)" <gi...@apache.org>]

tzulitai commented on code in PR #22150:
URL: https://github.com/apache/flink/pull/22150#discussion_r1149544856


##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -89,7 +90,7 @@
     private final KafkaRecordSerializationSchema<IN> recordSerializer;
     private final Callback deliveryCallback;
     private final KafkaRecordSerializationSchema.KafkaSinkContext kafkaSinkContext;
-
+    private final AtomicReference<Exception> asyncProducerException = new AtomicReference<>();

Review Comment:
   The fix overall makes sense to me, although I'm wondering if we can avoid using an `AtomicReference` and just use a `volatile` reference instead? That would be better for overhead reasons because this reference lookup happens on the critical path on every `write()`. Using `AtomicReference` you essentially have this extra pointer chase to the wrapped `volatile` variable.
   
   It looks like the only functionality that requires the solution to use an `AtomicReference` is the `compareAndSet` method in the callback, to make sure that we only ever update the reference once. I think relaxing this is ok. wdyt @mas-chen?



-- 
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