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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
