You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Jingsong Lee (Jira)" <ji...@apache.org> on 2020/02/26 02:50:00 UTC
[jira] [Comment Edited] (FLINK-16281) parameter 'maxRetryTimes' can
not work in JDBCUpsertTableSink
[ https://issues.apache.org/jira/browse/FLINK-16281?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17045097#comment-17045097 ]
Jingsong Lee edited comment on FLINK-16281 at 2/26/20 2:49 AM:
---------------------------------------------------------------
I got it, More specifically, it is bug in {{AppendOnlyWriter}}, {{UpsertWriter}} is OK, right?
was (Author: lzljs3620320):
I got it, More specifically, it is bug in {{AppendOnlyWriter, }}UpsertWriter is OK, right?
> parameter 'maxRetryTimes' can not work in JDBCUpsertTableSink
> -------------------------------------------------------------
>
> Key: FLINK-16281
> URL: https://issues.apache.org/jira/browse/FLINK-16281
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Ecosystem
> Affects Versions: 1.10.0
> Reporter: Leonard Xu
> Assignee: Leonard Xu
> Priority: Major
> Fix For: 1.10.1, 1.11.0
>
>
> When I insert data to a mysql table that do no exists in my test database will get exception,
> {code:java}
> Caused by: com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Table 'test.gmv_table' doesn't existCaused by: com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: Table 'test.gmv_table' doesn't exist 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:425) at com.mysql.jdbc.Util.getInstance(Util.java:408) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:944) at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3933) at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3869) at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2524) at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2675) at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2465) at com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:1912) at com.mysql.jdbc.PreparedStatement.executeUpdateInternal(PreparedStatement.java:2133) at com.mysql.jdbc.PreparedStatement.executeBatchSerially(PreparedStatement.java:1810) ... 44 more
> {code}
> but after I increased the 'connector.write.max-retries' value from 1 to 3, the exception disappeared. :(
> So, I look up the flush implement code :
> {code:java}
> public synchronized void flush() throws Exception {
> checkFlushException();
> for (int i = 1; i <= maxRetryTimes; i++) {
> try {
> jdbcWriter.executeBatch();
> batchCount = 0;
> break;
> } catch (SQLException e) {
> LOG.error("JDBC executeBatch error, retry times = {}", i, e);
> if (i >= maxRetryTimes) {
> throw e;
> }
> Thread.sleep(1000 * i);
> }
> }
> }{code}
> I found the `jdbcWriter` will clear its `batchedArgs` member after first call `jdbcWriter.executeBatch()` as follows:
> {code:java}
> //com.mysql.jdbc.PreparedStatement
> finally {
> this.statementExecuting.set(false);
> clearBatch();
> }
> // clearBatch() function implement
> public void clearBatch() throws SQLException {
> synchronized (checkClosed().getConnectionMutex()) {
> if (this.batchedArgs != null) {
> this.batchedArgs.clear();
> }
> }
> }
> {code}
> and the next time（ where i> 1） to call `jdbcWriter.executeBatch()` , the function will return empty array rather than execute the flush data
> {code:java}
> //com.mysql.jdbc.PreparedStatement
> if (this.batchedArgs == null || this.batchedArgs.size() == 0) {
> return new long[0];
> }
> ... // flush data code{code}
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)