You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2023/02/02 03:40:32 UTC
[inlong] 01/04: [INLONG-7291][Sort] Fix bug of dirty sink not opening for jdbc multiple sink (#7288)
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 <32...@users.noreply.github.com>
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);
+ }
}
}