[ https://issues.apache.org/jira/browse/FLINK-10455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16673230#comment-16673230 ]
ASF GitHub Bot commented on FLINK-10455: ---------------------------------------- pnowojski commented on a change in pull request #6989: [FLINK-10455][Kafka Tx] Close transactional producers in case of failure and termination URL: https://github.com/apache/flink/pull/6989#discussion_r230404337 ########## File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java ########## @@ -150,6 +154,13 @@ protected TXN currentTransaction() { return currentTransactionHolder == null ? null : currentTransactionHolder.handle; } + @Nonnull + protected Map<Long, TXN> pendingTransactions() { + return pendingCommitTransactions.entrySet().stream() + .map(e -> new AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().handle)) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); Review comment: why do you `collect` and not just return a stream? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Potential Kafka producer leak in case of failures > ------------------------------------------------- > > Key: FLINK-10455 > URL: https://issues.apache.org/jira/browse/FLINK-10455 > Project: Flink > Issue Type: Bug > Components: Kafka Connector > Affects Versions: 1.5.2 > Reporter: Nico Kruber > Assignee: Andrey Zagrebin > Priority: Major > Labels: pull-request-available > > If the Kafka brokers' timeout is too low for our checkpoint interval [1], we > may get an {{ProducerFencedException}}. Documentation around > {{ProducerFencedException}} explicitly states that we should close the > producer after encountering it. > By looking at the code, it doesn't seem like this is actually done in > {{FlinkKafkaProducer011}}. Also, in case one transaction's commit in > {{TwoPhaseCommitSinkFunction#notifyCheckpointComplete}} fails with an > exception, we don't clean up (nor try to commit) any other transaction. > -> from what I see, {{TwoPhaseCommitSinkFunction#notifyCheckpointComplete}} > simply iterates over the {{pendingCommitTransactions}} which is not touched > during {{close()}} > Now if we restart the failing job on the same Flink cluster, any resources > from the previous attempt will still linger around. > [1] > https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/connectors/kafka.html#kafka-011 -- This message was sent by Atlassian JIRA (v7.6.3#76005)