You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Piotr Nowojski (JIRA)" <ji...@apache.org> on 2018/05/02 15:17:00 UTC

[jira] [Commented] (FLINK-9287) KafkaProducer011 seems to leak threads when not in exactly-once mode

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

Piotr Nowojski commented on FLINK-9287:
---------------------------------------

Yes, indeed there is such bug. The problem boils down to {{org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011#abort}} being used for both aborting transactions and closing them at the end of the operator's lifecycle. For EXACTLY_ONCE in both cases we should be closing {{FlinkKafkaProducer}} but for AT_LEAST_ONCE and NONE we should be closing them only during operator close - which is not happening.

I'm not sure, but maybe we should split EXACTLY_ONCE and NON-EXACTLY_ONCE implementations, instead of trying to squeeze in both of them at the same time into one class. 

CC:
[~tzulitai]  [~gjy]

> KafkaProducer011 seems to leak threads when not in exactly-once mode
> --------------------------------------------------------------------
>
>                 Key: FLINK-9287
>                 URL: https://issues.apache.org/jira/browse/FLINK-9287
>             Project: Flink
>          Issue Type: Bug
>          Components: Kafka Connector
>    Affects Versions: 1.4.2
>            Reporter: Christopher Ng
>            Priority: Minor
>
> {{KafkaProducer011}} appears to be leaking {{kafka-producer-network-thread}} threads.  As far as I can tell it happens when it is not in EXACTLY_ONCE mode, it seems that it creates a {{FlinkKafkaProducer}} but never closes it, even when the {{FlinkKafkaProducer011}} itself is closed.
> I observed this when running a local cluster and submitting and then cancelling a job, a lot of kafka threads were left alive afterwards.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)