[jira] [Commented] (KAFKA-4767) KafkaProducer is not joining its IO thread properly

["Colin P. McCabe (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-4767?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15894818#comment-15894818 ] 

Colin P. McCabe commented on KAFKA-4767:
----------------------------------------

I agree that the interrupt state of the thread should be preserved.  In other words, we should invoke {{Thread.currentThread().interrupt()}} in the catch block for {{InterruptedException}}.  However, the current behavior of aborting the thread join when an {{InterruptedException}} is received seems correct to me.

bq. I understand that it will eventually go away. But that does not cut it for us. We would like the IO thread to be shutdown after close returns. And that does not happen if we get an interrupt during close().

Fundamentally, if you interrupt an operation, you are not waiting for it to complete.  That's what interruption means.

> 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.10.2.0
>            Reporter: Buğra Gedik
>            Assignee: huxi
>            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)