[
https://issues.apache.org/jira/browse/FLINK-16281?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Leonard Xu updated FLINK-16281:
-------------------------------
Description:
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}
was:
parameter 'maxRetryTimes' can not work in JDBCUpsertTableSink,
because PreparedStatement will clear batchedArgs info after first
call function executeBatchInternal() 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 to call function executeBatchInternal() can not go on because
this piece code:
{code:java}
//com.mysql.jdbc.PreparedStatement
if (this.batchedArgs == null || this.batchedArgs.size() == 0) {
return new long[0];
}
{code}
> 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
> 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)
