MINOR: improve RecordCollectorImpl #17185
Conversation
| @@ -81,7 +81,7 @@ public class RecordCollectorImpl implements RecordCollector { | |||
| private final Sensor droppedRecordsSensor; | |||
| private final Map<String, Sensor> producedSensorByTopic = new HashMap<>(); | |||
|
|
|||
| private final AtomicReference<KafkaException> sendException; | |||
| private final AtomicReference<KafkaException> sendException = new AtomicReference<>(null); | |||
There was a problem hiding this comment.
There is no reason to get this from StreamsProducer so pulling it into this class.
| @@ -529,7 +528,7 @@ private void checkForException() { | |||
| final KafkaException exception = sendException.get(); | |||
|
|
|||
| if (exception != null) { | |||
| sendException.set(null); | |||
| sendException.compareAndSet(exception, null); | |||
There was a problem hiding this comment.
I believe this is actually fixing a race condition... with set() we might overwrite (and thus drop) and exception which is set after we called get() (even if I am not 100% sure how bad it is if we drop it, give that we throw exception; below anyway...?
There was a problem hiding this comment.
Yeah... not sure that part makes a difference. But I guess there is nothing wrong with it?
The rest of PR makes sense to me
|
After re-thinking, the test we remove because it failed actually exposed that the original idea of this PR is incorrect. We need to keep |
mjsax
force-pushed
the
minor-recordcollectorimpl
branch
from
2d4277e
to
74f1042
Compare
Found this while working on #17169, but did not want to piggy-back this change to the other PR but isolate as it's unrelated cleanup.