You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@seatunnel.apache.org by ga...@apache.org on 2023/01/06 03:01:21 UTC
[incubator-seatunnel] branch dev updated: [Improve][JDBC] improve jdbc sink option (#3864)
This is an automated email from the ASF dual-hosted git repository.
gaojun2048 pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 768a9300e [Improve][JDBC] improve jdbc sink option (#3864)
768a9300e is described below
commit 768a9300e8f6f01e4e04a27f684ce6ca15265639
Author: Eric <ga...@gmail.com>
AuthorDate: Fri Jan 6 11:01:16 2023 +0800
[Improve][JDBC] improve jdbc sink option (#3864)
* improve jdbc sink option
* improve option default get
---
docs/en/connector-v2/sink/Jdbc.md | 2 +-
.../seatunnel/jdbc/config/JdbcConfig.java | 11 ++++-----
.../internal/options/JdbcConnectionOptions.java | 27 ++++++++--------------
3 files changed, 16 insertions(+), 24 deletions(-)
diff --git a/docs/en/connector-v2/sink/Jdbc.md b/docs/en/connector-v2/sink/Jdbc.md
index ec31ce87d..c01b173b2 100644
--- a/docs/en/connector-v2/sink/Jdbc.md
+++ b/docs/en/connector-v2/sink/Jdbc.md
@@ -38,7 +38,7 @@ support `Xa transactions`. You can set `is_exactly_once=true` to enable it.
| support_upsert_by_query_primary_key_exist | Boolean | No | false |
| connection_check_timeout_sec | Int | No | 30 |
| max_retries | Int | No | 3 |
-| batch_size | Int | No | 300 |
+| batch_size | Int | No | 1000 |
| batch_interval_ms | Int | No | 1000 |
| is_exactly_once | Boolean | No | false |
| xa_data_source_class_name | String | No | - |
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcConfig.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcConfig.java
index b1d395abb..72efa6fc8 100644
--- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcConfig.java
+++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcConfig.java
@@ -26,6 +26,7 @@ import org.apache.seatunnel.shade.com.typesafe.config.Config;
import java.io.Serializable;
import java.util.List;
+@SuppressWarnings("checkstyle:MagicNumber")
public class JdbcConfig implements Serializable {
private static final int DEFAULT_CONNECTION_CHECK_TIMEOUT_SEC = 30;
private static final boolean DEFAULT_AUTO_COMMIT = true;
@@ -46,22 +47,20 @@ public class JdbcConfig implements Serializable {
public static final Option<Boolean> AUTO_COMMIT = Options.key("auto_commit").booleanType().defaultValue(DEFAULT_AUTO_COMMIT).withDescription("auto commit");
- public static final Option<Integer> BATCH_SIZE = Options.key("batch_size").intType().noDefaultValue().withDescription("batch size");
+ public static final Option<Integer> BATCH_SIZE = Options.key("batch_size").intType().defaultValue(1000).withDescription("batch size");
public static final Option<Integer> FETCH_SIZE = Options.key("fetch_size").intType().defaultValue(0).withDescription("For queries that return a large number of objects, " +
"you can configure the row fetch size used in the query to improve performance by reducing the number database hits required to satisfy the selection criteria. Zero means use jdbc default value.");
- public static final Option<Integer> BATCH_INTERVAL_MS = Options.key("batch_interval_ms").intType().noDefaultValue().withDescription("batch interval milliSecond");
-
+ public static final Option<Integer> BATCH_INTERVAL_MS = Options.key("batch_interval_ms").intType().defaultValue(1000).withDescription("batch interval milliSecond");
public static final Option<Boolean> IS_EXACTLY_ONCE = Options.key("is_exactly_once").booleanType().defaultValue(true).withDescription("exactly once");
public static final Option<String> XA_DATA_SOURCE_CLASS_NAME = Options.key("xa_data_source_class_name").stringType().noDefaultValue().withDescription("data source class name");
+ public static final Option<Integer> MAX_COMMIT_ATTEMPTS = Options.key("max_commit_attempts").intType().defaultValue(3).withDescription("max commit attempts");
- public static final Option<String> MAX_COMMIT_ATTEMPTS = Options.key("max_commit_attempts").stringType().noDefaultValue().withDescription("max commit attempts");
-
- public static final Option<String> TRANSACTION_TIMEOUT_SEC = Options.key("transaction_timeout_sec").stringType().noDefaultValue().withDescription("transaction timeout (second)");
+ public static final Option<Integer> TRANSACTION_TIMEOUT_SEC = Options.key("transaction_timeout_sec").intType().defaultValue(-1).withDescription("transaction timeout (second)");
public static final Option<String> TABLE = Options.key("table").stringType().noDefaultValue().withDescription("table");
diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/options/JdbcConnectionOptions.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/options/JdbcConnectionOptions.java
index c10cbf96e..540f54ec9 100644
--- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/options/JdbcConnectionOptions.java
+++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/options/JdbcConnectionOptions.java
@@ -26,31 +26,24 @@ public class JdbcConnectionOptions
implements Serializable {
private static final long serialVersionUID = 1L;
-
- private static final int DEFAULT_MAX_RETRIES = 3;
- private static final int DEFAULT_BATCH_SIZE = 300;
- private static final int DEFAULT_BATCH_INTERVAL_MS = 1000;
- private static final int DEFAULT_MAX_COMMIT_ATTEMPTS = 3;
- private static final int DEFAULT_TRANSACTION_TIMEOUT_SEC = -1;
-
public String url;
public String driverName;
public int connectionCheckTimeoutSeconds = JdbcConfig.CONNECTION_CHECK_TIMEOUT_SEC.defaultValue();
- public int maxRetries = DEFAULT_MAX_RETRIES;
+ public int maxRetries = JdbcConfig.MAX_RETRIES.defaultValue();
public String username;
public String password;
public String query;
public boolean autoCommit = JdbcConfig.AUTO_COMMIT.defaultValue();
- public int batchSize = DEFAULT_BATCH_SIZE;
- public int batchIntervalMs = DEFAULT_BATCH_INTERVAL_MS;
+ public int batchSize = JdbcConfig.BATCH_SIZE.defaultValue();
+ public int batchIntervalMs = JdbcConfig.BATCH_INTERVAL_MS.defaultValue();
public String xaDataSourceClassName;
- public int maxCommitAttempts = DEFAULT_MAX_COMMIT_ATTEMPTS;
+ public int maxCommitAttempts = JdbcConfig.MAX_COMMIT_ATTEMPTS.defaultValue();
- public int transactionTimeoutSec = DEFAULT_TRANSACTION_TIMEOUT_SEC;
+ public int transactionTimeoutSec = JdbcConfig.TRANSACTION_TIMEOUT_SEC.defaultValue();
public JdbcConnectionOptions() {
}
@@ -111,16 +104,16 @@ public class JdbcConnectionOptions
private String url;
private String driverName;
private int connectionCheckTimeoutSeconds = JdbcConfig.CONNECTION_CHECK_TIMEOUT_SEC.defaultValue();
- private int maxRetries = DEFAULT_MAX_RETRIES;
+ private int maxRetries = JdbcConfig.MAX_RETRIES.defaultValue();
private String username;
private String password;
private String query;
private boolean autoCommit = JdbcConfig.AUTO_COMMIT.defaultValue();
- private int batchSize = DEFAULT_BATCH_SIZE;
- private int batchIntervalMs = DEFAULT_BATCH_INTERVAL_MS;
+ private int batchSize = JdbcConfig.BATCH_SIZE.defaultValue();
+ private int batchIntervalMs = JdbcConfig.BATCH_INTERVAL_MS.defaultValue();
private String xaDataSourceClassName;
- private int maxCommitAttempts = DEFAULT_MAX_COMMIT_ATTEMPTS;
- private int transactionTimeoutSec = DEFAULT_TRANSACTION_TIMEOUT_SEC;
+ private int maxCommitAttempts = JdbcConfig.MAX_COMMIT_ATTEMPTS.defaultValue();
+ private int transactionTimeoutSec = JdbcConfig.TRANSACTION_TIMEOUT_SEC.defaultValue();
private JdbcConnectionOptionsBuilder() {
}