Mailing-List: contact dev-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@kafka.apache.org
Date: Thu, 16 Feb 2017 14:41:41 +0000 (UTC)
From: "huxi (JIRA)" <jira@apache.org>
To: dev@kafka.apache.org
Message-ID: <JIRA.13043239.1487160599000.99734.1487256101880@Atlassian.JIRA>
In-Reply-To: <JIRA.13043239.1487160599000@Atlassian.JIRA>
References: <JIRA.13043239.1487160599000@Atlassian.JIRA> <JIRA.13043239.1487160599492@jira-lw-us.apache.org>
Subject: [jira] [Commented] (KAFKA-4767) KafkaProducer is not joining its IO
 thread properly
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
archived-at: Thu, 16 Feb 2017 14:41:47 -0000


    [ https://issues.apache.org/jira/browse/KAFKA-4767?page=3Dcom.atlassian=
.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=3D1587=
0055#comment-15870055 ]=20

huxi commented on KAFKA-4767:
-----------------------------

What do you mean by "leaking the IO thread"? Do you mean it could not be sh=
ut down successfully after interrupting the user thread in which KafkaProdu=
cer.close was invoked?  This should be not gonna happen since this.sender.i=
nitiateClose() would always be run even when you interrupt the user thread.=
=20

In my opinion, interrupting the user thread is no different from invoking i=
oThread.join with a relatively small timeout because there is still a chanc=
e to force close the IO thread and wait it again. That's also why we swallo=
w InterruptedException during the first join.=20

Does it look good to you though? And for sake of the curiosity, did you enc=
ounter any cases where IO thread got failed to be shut down?

> 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=20
>    Affects Versions: 0.11.0.0
>            Reporter: Bu=C4=9Fra 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 leav=
ing 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 {=20
>          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)