Hi Zhong Tang, 我查看了该jira有关的重连pr，https://github.com/apache/flink/pull/8429 ，但该pr主要通过重连机制来规避网络问题导致的连接失效， 但由于我的业务场景数据比较稀疏，遭遇wait timeout连接失效是一个常见的情况，有最大次数限制的重连机制并不是很适合。
主要的需求其实是connection的持久保活。 Thanks, Xuhui Mao Zhonghan Tang <13122260...@163.com> 于2020年6月30日周二 下午12:05写道： > 可以参考这个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 >
