wuchong commented on a change in pull request #12427:
URL: https://github.com/apache/flink/pull/12427#discussion_r434371368
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/InsertOrUpdateJdbcExecutor.java
##########
@@ -83,6 +88,21 @@ public void open(Connection connection) throws SQLException {
updateStatement = connection.prepareStatement(updateSQL);
}
+ @Override
+ public void reopen(Connection connection) throws SQLException {
+ try {
+ existStatement.close();
+ insertStatement.close();
+ updateStatement.close();
+ } catch (SQLException e) {
+ LOG.info("PreparedStatement close failed.", e);
+ }
+
+ existStatement = connection.prepareStatement(existSQL);
+ insertStatement = connection.prepareStatement(insertSQL);
+ updateStatement = connection.prepareStatement(updateSQL);
Review comment:
I think we don't need to introduce a new `reopen` interface, the
implementation is always the combination of `close` and `open`. You can just
call `open` and then `close`.
But maybe we need to move the construction of `batch` map into constuctor.
This is also better because we can mark `batch` as `final` then. We can also
rename `open` to `openConnection` and `close` to `closeConnection` to make the
behavior more explicitly.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
##########
@@ -175,6 +176,15 @@ public synchronized void flush() throws IOException {
if (i >= executionOptions.getMaxRetries()) {
throw new IOException(e);
}
+ try {
+ if
(!connection.isValid(JdbcLookupFunction.CONNECTION_CHECK_TIMEOUT)) {
+ connection =
connectionProvider.reestablishConnection();
+
jdbcStatementExecutor.reopen(connection);
+ }
+ } catch (Exception excpetion) {
+ LOG.error("JDBC connection is not
valid, and reestablish connection failed.", excpetion);
+ throw new RuntimeException("Reestablish
JDBC connection failed", excpetion);
Review comment:
Use `IOException` instead.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
