Cause: java.sql.SQLException: No operations allowed after statement closed.

Posted 2023-03-01 augie_ly

Cause: java.sql.SQLException: No operations allowed after statement closed.

需求背景：多个大表关联打成一个大宽表，数据量是千万级。

一开始直接写Flink SQL非常慢并且严重反压，运行一段时间会报binlog过期的问题。

2022-11-06 16:50:34
java.lang.RuntimeException: One or more fetchers have encountered exception
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:225)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:185)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143)
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:526)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:811)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:760)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:954)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:933)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568)
	at java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:150)
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
	... 1 more
Caused by: java.lang.IllegalStateException: The connector is trying to read binlog starting at Structversion=1.6.4.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1667724634615,db=,server_id=0,file=mysql-bin.000970,pos=8325121,row=0, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed.
	at com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext.loadStartingOffsetState(StatefulTaskContext.java:190)
	at com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext.configure(StatefulTaskContext.java:116)
	at com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:99)
	at com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:67)
	at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.checkSplitOrStartNext(MySqlSplitReader.java:159)
	at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.fetch(MySqlSplitReader.java:71)
	at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)
	... 6 more

现在用jar包方式, 我用的flink（1.13.3）+ mybatis（3.5.3）+ druid（1.2.5）连接池,
FlinkCDC读主表，其他需要的字段直接查数据库。也是非常慢，运行一段时间报连接数据库超时的问题：

我的mysql的wait_timout设置的是7200秒,druid连接池的配置如下:

<property name="keepAlive" value="true" />
<property name="maxActive" value="20" />
<property name="maxWait" value="60000" />
<property name="validationQuery" value="select 1" />
<property name="minEvictableIdleTimeMillis" value="300000" />
<property name="timeBetweenEvictionRunsMillis" value="60000" />
<property name="minIdle" value="5" />
<property name="testWhileIdle" value="true" />
<property name="poolPreparedStatements" value="true" />
<property name="maxPoolPreparedStatementPerConnectionSize" value="5" />
<property name="queryTimeout" value="388000" />

后面打开removeAbandoned功能进行测试，添加druid连接池配置：

<property name="removeAbandoned" value="true" /> <!-- 打开removeAbandoned功能 -->
<property name="removeAbandonedTimeout" value="388000" /> <!-- 388000秒，也就是4.5天 -->
<property name="logAbandoned" value="true" /> <!-- 关闭abanded连接时输出错误日志 -->

报错如下：

Cause: java.sql.SQLException: No operations allowed after statement closed.

不知道是不是配置哪里有问题，或者不知道还有没有别的解决方案。欢迎各位大佬指点讨论。