[FLINK-30371][Connector/JDBC] Fix the problem of JdbcOutputFormat database connection leak

[EchoLee5]

Fix the problem of JdbcOutputFormat database connection leak

[boring-cyborg[bot]]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[EchoLee5]

@EchoLee5 Thanks for the PR, is it possible to add a test for this?

@MartijnVisser I have added test cases, please help me review the code again, thank you.

[MartijnVisser]

@EchoLee5 Can you please rebase and use the new testing setup?

[KevDi]

@MartijnVisser is there any update on this PR? I can also open a new PR with the fixes from @EchoLee5.

[tamirsagi]

connectionProvider is configured as final and is AutoClossable which can be used in try(connectionProvider) which handles the closeConnection automatically.

[KevDi]

@tamirsagi Is this fix still needed? Or the one i send in? #75 ?

[tamirsagi]

@tamirsagi Is this fix still needed? Or the one i send in? #75 ?

if you are asking whether its still relevant, then yes. yours is similar. yet, I'd wrap connectionProvider with try after calling #flush because it's cleaner & safer. + I don't see any reason to store an exception to flushException during #close.

[KevDi]

@tamirsagi it is basicly the same fix i asked on the Jira Board and they said i should open up a new PR thats why there are two :) what exactly do you mean by wrapping it? Do you have some example code?

[tamirsagi]

@tamirsagi it is basicly the same fix i asked on the Jira Board and they said i should open up a new PR thats why there are two :) what exactly do you mean by wrapping it? Do you have some example code?

JdbcOutputFormat#close method add it there

try (connectionProvider) {
  if (jdbcStatementExecutor != null) {
       jdbcStatementExecutor.closeStatements();
    }
  } catch (SQLException e) {
     LOG.warn("Close JDBC writer failed.", e);
  }

[simon-paradis-jive]

Would that get eventually released in the x.y.z-1.19 series ?

After a database reboot, our Flink job took several minutes to recover from the dead connection. I think this could fix it.

Flink 1.20 has also been out for a while, will we have a -1.20 matching serie ? Would it be safe to run the -1.19 jdbc connector serie against Flink 1.20 ? For now we're staying on 1.19