[ https://issues.apache.org/jira/browse/KAFKA-4767?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15869912#comment-15869912 ]
Buğra Gedik commented on KAFKA-4767: ------------------------------------ I think we are getting close to understanding each other. Here is my problem: I have code running in a thread, which calls KafkaProducer.close(). If the thread happens to get an interrupt while it is inside ``KafkaProducer.close()`` it ends up leaking the Kafka IO thread. This happens because join() inside close() ends up getting inerrupted. The way we join threads in our Java code is as follows: {code} try { myThread.join(); } catch(final InterruptedException e) { // some call to tell myThread to finish up immediately myThread.interrupt(); // it could be some other call for your case try { myThread.join(); } catch(final InterruptedException e) { // could be something else, such as logging in your case throw new MyCustomException(..., e); } finally { // make sure we maintain the interrupted status Thread.currentThread.interrupt(); } } catch (final ) {code} > KafkaProducer is not joining its IO thread properly > --------------------------------------------------- > > Key: KAFKA-4767 > URL: https://issues.apache.org/jira/browse/KAFKA-4767 > Project: Kafka > Issue Type: Bug > Components: producer > Affects Versions: 0.11.0.0 > Reporter: Buğra Gedik > Priority: Minor > > The {{KafkaProducer}} is not properly joining the thread it creates. The code > is like this: > {code} > try { > this.ioThread.join(timeUnit.toMillis(timeout)); > } catch (InterruptedException t) { > firstException.compareAndSet(null, t); > log.error("Interrupted while joining ioThread", t); > } > {code} > If the code is interrupted while performing the join, it will end up leaving > the io thread running. The correct way of handling this is a follows: > {code} > try { > this.ioThread.join(timeUnit.toMillis(timeout)); > } catch (InterruptedException t) { > // propagate the interrupt > this.ioThread.interrupt(); > try { > this.ioThread.join(); > } catch (InterruptedException t) { > firstException.compareAndSet(null, t); > log.error("Interrupted while joining ioThread", t); > } finally { > // make sure we maintain the interrupted status > Thread.currentThread.interrupt(); > } > } > {code} -- This message was sent by Atlassian JIRA (v6.3.15#6346)