Flink在下游算子没有idle的情况下自动产生背压

[carryxyh <ca...@gmail.com>]

flink 版本1.13.0，source算子均为mysql-cdc 1.4，使用flink sql


全都阻塞在获取segment上（Localbufferpool.requestMemorySegmentBlocking）


outpoolusage指标，有的背压算子是一直是0，有的是一直是1。


下游算子一直是idle状态，在等待上游数据。阻塞在mailbox上


mysql-cdc在运行过程中出现了一些异常：


ERROR io.debezium.connector.mysql.SnapshotReader:219 - Fa
iled due to error: Aborting snapshot due to error when last running 'SELECT * FROM `db`.`table1`': Streaming result set com.mysql.cj
.protocol.a.result.ResultsetRowsStreaming@45edf631 is still active. No statements may be issued when any streaming result sets are open 
and in use on a given connection. Ensure that you have called .close() on any active streaming result sets before attempting more querie
s.
org.apache.kafka.connect.errors.ConnectException: Streaming result set com.mysql.cj.protocol.a.result.ResultsetRowsStreaming@45edf631 is
 still active. No statements may be issued when any streaming result sets are open and in use on a given connection. Ensure that you hav
e called .close() on any active streaming result sets before attempting more queries. Error code: 0; SQLSTATE: S1000.
        at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241)
        at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:218)
        at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:857)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLException: Streaming result set com.mysql.cj.protocol.a.result.ResultsetRowsStreaming@45edf631 is still active. N
o statements may be issued when any streaming result sets are open and in use on a given connection. Ensure that you have called .close(
) on any active streaming result sets before attempting more queries.
        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:129)
        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:97)
        at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
        at com.mysql.cj.jdbc.ConnectionImpl.commit(ConnectionImpl.java:814)
        at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:773)
        ... 3 common frames omitted


但是任务没有发生restarting，一直在运行。


感觉上像是发生了泄露，不知道有什么排查思路么？