This is an automated email from the ASF dual-hosted git repository. healchow pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/inlong.git
commit e8432ed8838d7c121c3b0fcf17f6a05859d4a005 Author: Yizhou Yang <[email protected]> AuthorDate: Thu Feb 2 10:03:54 2023 +0800 [INLONG-7291][Sort] Fix bug of dirty sink not opening for jdbc multiple sink (#7288) --- .../inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java b/inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java index 67e9adac0..ead5b58a3 100644 --- a/inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java +++ b/inlong-sort/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java @@ -24,6 +24,7 @@ import org.apache.flink.api.common.state.ListState; import org.apache.flink.api.common.state.ListStateDescriptor; import org.apache.flink.api.common.typeinfo.TypeHint; import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.jdbc.JdbcExecutionOptions; import org.apache.flink.connector.jdbc.internal.connection.JdbcConnectionProvider; import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider; @@ -219,6 +220,11 @@ public class JdbcMultiBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatc executionOptions.getBatchIntervalMs(), executionOptions.getBatchIntervalMs(), TimeUnit.MILLISECONDS); + try { + dirtySinkHelper.open(new Configuration()); + } catch (Exception e) { + throw new IOException(e); + } } }
