可以参考这个jira https://issues.apache.org/jira/browse/FLINK-12494 1. Throw execption and let flink runtime handle it; 2. Handle it in OutputFormat;
| | Zhonghan Tang | | 13122260...@163.com | 签名由网易邮箱大师定制 On 06/30/2020 11:53，shizk233<wangwangdaxian...@gmail.com> wrote： Hi All, 最近使用flink处理数据写入mysql sink，但由于业务场景在晚上没有数据流入，会触发mysql wait timeout限制（默认的8小时）导致连接失效。 即使在mysql url中添加了autoReconnect=true参数，仍会产生相应的异常，具体信息见下。 版本信息： flink 1.10.1 mysql server 5.6.47 mysql Connector/J 5.1.49 请问： 1.flink的jdbc connector是否可以采用连接池模型？如果只使用一个connection，是否可以添加某种心跳机制以保持active？ 2.连接失效后是否有可能丢数（因为源码没找到存储record的state field）？ 3.在当前版本下，连接失效有什么比较好的解决方案吗？ Thanks, Xuhui Mao 异常信息： 2020-06-24 22:39:46,923 ERROR org.apache.flink.api.java.io.jdbc.JDBCUpsertOutputFormat - JDBC executeBatch error, retry times = 1 java.sql.SQLException: Could not retrieve transaction read-only status from server at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:965) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:898) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:887) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:861) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:878) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:874) at com.mysql.jdbc.ConnectionImpl.isReadOnly(ConnectionImpl.java:3523) at com.mysql.jdbc.ConnectionImpl.isReadOnly(ConnectionImpl.java:3490) at com.mysql.jdbc.PreparedStatement.executeBatchInternal(PreparedStatement.java:1287) at com.mysql.jdbc.StatementImpl.executeBatch(StatementImpl.java:954) at org.apache.flink.api.java.io.jdbc.writer.AppendOnlyWriter.executeBatch(AppendOnlyWriter.java:70) at org.apache.flink.api.java.io.jdbc.JDBCUpsertOutputFormat.flush(JDBCUpsertOutputFormat.java:161) at org.apache.flink.api.java.io.jdbc.JDBCUpsertOutputFormat.lambda$open$0(JDBCUpsertOutputFormat.java:125) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure The last packet successfully received from the server was 10,384,059 milliseconds ago. The last packet sent successfully to the server was 10,384,063 milliseconds ago. at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at com.mysql.jdbc.Util.handleNewInstance(Util.java:403) at com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:990) at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3706) at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2506) at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2675) at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2465) at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2439) at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1365) at com.mysql.jdbc.ConnectionImpl.isReadOnly(ConnectionImpl.java:3515) ... 13 more Caused by: java.net.SocketException: Connection reset at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:115) at java.net.SocketOutputStream.write(SocketOutputStream.java:155) at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3688) ... 19 more
