[GitHub] [flink] wuchong commented on a change in pull request #12427: [FLINK-16681][jdbc] Fix the bug that jdbc lost connection after a lon…

[GitBox <gi...@apache.org>]

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:
users@infra.apache.org