[FLINK-23182] Connection leak in RMQSource

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.13.1, 1.12.4
• Fix Version/s: 1.14.0, 1.12.5, 1.13.2
• Component/s: Connectors/ RabbitMQ
• Labels: pull-request-available
• Assignee: Michał Ciesielczyk
• Reporter: Michał Ciesielczyk
• Created: 29/Jun/21 11:27
• Updated: 17/Nov/21 15:21
• Resolved: 06/Jul/21 16:46

Description

The RabbitMQ connection is not closed properly in the RMQSource connector in case of failures. This leads to a connection leak (we loose handles to still opened connections) that will last until the Flink TaskManager is either stopped or crashes.

The issue is caused by improper resource releasing in open and close methods of RMQSource:

• https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java#L260 - here the connection is opened, but not closed in case of failure (e.g. caused by invalid queue configuration)
• https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java#L282 - here the connection might not closed properly if stopping the consumer causes a failure first

In both cases, the solution is relatively simple - make sure that the connection#close is always called if it should be (failing to close one resource should not prevent other close methods from being called). In open we probably can silently close allocated resources (as the process did not succeed eventually anyway). In close, we should either throw the first caught exception or the last one, and log all the others as warnings.

Issue Links

• is duplicated by: FLINK-20244 RMQSource does not ACK duplicated messages
• links to: GitHub Pull Request #16333