SAMZA-1539: KafkaProducer potential hang on close() when task.drop.pr… #390
Conversation
…oducer.errors==true
|
@prateekm @nickpan47 @vjagadish1989 for review |
| if (isFatalException) { | ||
| fatalException.compareAndSet(null, producerException) | ||
| // Prevent each callback from closing and nulling producer for the same failure. | ||
| if (currentProducer == producerRef.get()) { |
There was a problem hiding this comment.
If both currentProducer and producerRef.get() are null here, it can have null pointer exception. I would recommend to have a null check to make the code more robust.
There was a problem hiding this comment.
- The java doc specifies that currentProducer must not be null
- If currentProducer was null, the right thing to do is throw an exception because we can't properly handle the exception, so then the debate is NPE or other exception. I don't see the value.
| if (producer != null) { | ||
| producer.close // Also performs the equivalent of a flush() | ||
| } | ||
| stopped = true |
There was a problem hiding this comment.
would it make more sense to put stopped = true after stop is done?
There was a problem hiding this comment.
No. The flag is used to prevent subsequent producer creations. That should happen immediately when stop() is invoked.
There was a problem hiding this comment.
Yeah, the logic makes sense. It might be better to call it stopTriggered or something else instead of stopped, to avoid the confusion, but that's minor and optional.
| if (currentProducer == null) { | ||
| if (dropProducerExceptions) { | ||
| // No producer to flush, but we're ignoring exceptions so just return. |
There was a problem hiding this comment.
Could be useful. Added it.
…oducer.errors==true Author: Jacob Maes <jmakes@apache.org> Reviewers: Boris Shkolnik <boryas@apache.org> Closes apache#390 from jmakes/samza-1539
…oducer.errors==true