You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/12/25 02:23:45 UTC
[flink] branch master updated: [FLINK-19691][jdbc] Support
connection.max-retry-timeout configuration for JDBC connector
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f5e829c [FLINK-19691][jdbc] Support connection.max-retry-timeout configuration for JDBC connector
f5e829c is described below
commit f5e829c4ffed88e1a79b76e9fc8579013502ddc0
Author: 肖佳文 <xi...@xiaomi.com>
AuthorDate: Tue Dec 15 16:59:04 2020 +0800
[FLINK-19691][jdbc] Support connection.max-retry-timeout configuration for JDBC connector
This closes #14387
---
docs/dev/table/connect.md | 3 +++
docs/dev/table/connect.zh.md | 3 +++
docs/dev/table/connectors/jdbc.md | 7 +++++++
docs/dev/table/connectors/jdbc.zh.md | 7 +++++++
.../connector/jdbc/JdbcConnectionOptions.java | 21 +++++++++++++++++--
.../jdbc/internal/JdbcBatchingOutputFormat.java | 3 +--
.../connection/JdbcConnectionProvider.java | 2 ++
.../connection/SimpleJdbcConnectionProvider.java | 5 +++++
.../jdbc/internal/options/JdbcOptions.java | 24 ++++++++++++++--------
.../jdbc/table/JdbcDynamicTableFactory.java | 16 ++++++++++++++-
.../connector/jdbc/table/JdbcLookupFunction.java | 5 +++--
.../jdbc/table/JdbcRowDataLookupFunction.java | 5 +++--
.../jdbc/table/JdbcTableSourceSinkFactory.java | 4 ++++
.../flink/table/descriptors/JdbcValidator.java | 2 ++
.../jdbc/table/JdbcDynamicTableFactoryTest.java | 14 +++++++++++++
.../jdbc/table/JdbcTableSourceSinkFactoryTest.java | 13 ++++++++++++
16 files changed, 117 insertions(+), 17 deletions(-)
diff --git a/docs/dev/table/connect.md b/docs/dev/table/connect.md
index c08006d..92e72d7 100644
--- a/docs/dev/table/connect.md
+++ b/docs/dev/table/connect.md
@@ -1301,6 +1301,9 @@ CREATE TABLE MyUserTable (
'connector.username' = 'name',
'connector.password' = 'password',
+ -- optional: jdbc connection max-retry-timeout
+ 'connector.connection.max-retry-timeout' = '60s',
+
-- **followings are scan options, optional, used when reading from a table**
-- optional: SQL query / prepared statement.
diff --git a/docs/dev/table/connect.zh.md b/docs/dev/table/connect.zh.md
index 3259882..444d772 100644
--- a/docs/dev/table/connect.zh.md
+++ b/docs/dev/table/connect.zh.md
@@ -1301,6 +1301,9 @@ CREATE TABLE MyUserTable (
'connector.username' = 'name',
'connector.password' = 'password',
+ -- optional: jdbc connection max-retry-timeout
+ 'connector.connection.max-retry-timeout' = '60s',
+
-- **followings are scan options, optional, used when reading from table**
-- These options must all be specified if any of them is specified. In addition,
diff --git a/docs/dev/table/connectors/jdbc.md b/docs/dev/table/connectors/jdbc.md
index 4aa7059..5c0a091 100644
--- a/docs/dev/table/connectors/jdbc.md
+++ b/docs/dev/table/connectors/jdbc.md
@@ -149,6 +149,13 @@ Connector Options
<td>The JDBC password.</td>
</tr>
<tr>
+ <td><h5>connection.max-retry-timeout</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">60s</td>
+ <td>Duration</td>
+ <td>Maximum timeout between retries. The timeout should be in second granularity and shouldn't be smaller than 1 second.</td>
+ </tr>
+ <tr>
<td><h5>scan.partition.column</h5></td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/docs/dev/table/connectors/jdbc.zh.md b/docs/dev/table/connectors/jdbc.zh.md
index 6e10074..b9d9044 100644
--- a/docs/dev/table/connectors/jdbc.zh.md
+++ b/docs/dev/table/connectors/jdbc.zh.md
@@ -149,6 +149,13 @@ Connector Options
<td>The JDBC password.</td>
</tr>
<tr>
+ <td><h5>connection.max-retry-timeout</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">60s</td>
+ <td>Duration</td>
+ <td>Maximum timeout between retries. The timeout should be in second granularity and shouldn't be smaller than 1 second.</td>
+ </tr>
+ <tr>
<td><h5>scan.partition.column</h5></td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcConnectionOptions.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcConnectionOptions.java
index 687c82e..761b4ab 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcConnectionOptions.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcConnectionOptions.java
@@ -35,16 +35,19 @@ public class JdbcConnectionOptions implements Serializable {
protected final String url;
protected final String driverName;
+ protected final int connectionCheckTimeoutSeconds;
@Nullable
protected final String username;
@Nullable
protected final String password;
- protected JdbcConnectionOptions(String url, String driverName, String username, String password) {
+ protected JdbcConnectionOptions(String url, String driverName, String username, String password, int connectionCheckTimeoutSeconds) {
+ Preconditions.checkArgument(connectionCheckTimeoutSeconds > 0);
this.url = Preconditions.checkNotNull(url, "jdbc url is empty");
this.driverName = Preconditions.checkNotNull(driverName, "driver name is empty");
this.username = username;
this.password = password;
+ this.connectionCheckTimeoutSeconds = connectionCheckTimeoutSeconds;
}
public String getDbURL() {
@@ -63,6 +66,10 @@ public class JdbcConnectionOptions implements Serializable {
return Optional.ofNullable(password);
}
+ public int getConnectionCheckTimeoutSeconds() {
+ return connectionCheckTimeoutSeconds;
+ }
+
/**
* Builder for {@link JdbcConnectionOptions}.
*/
@@ -71,6 +78,7 @@ public class JdbcConnectionOptions implements Serializable {
private String driverName;
private String username;
private String password;
+ private int connectionCheckTimeoutSeconds = 60;
public JdbcConnectionOptionsBuilder withUrl(String url) {
this.url = url;
@@ -92,8 +100,17 @@ public class JdbcConnectionOptions implements Serializable {
return this;
}
+ /**
+ * Set the maximum timeout between retries, default is 60 seconds.
+ * @param connectionCheckTimeoutSeconds the timeout seconds, shouldn't smaller than 1 second.
+ */
+ public JdbcConnectionOptionsBuilder withConnectionCheckTimeoutSeconds(int connectionCheckTimeoutSeconds) {
+ this.connectionCheckTimeoutSeconds = connectionCheckTimeoutSeconds;
+ return this;
+ }
+
public JdbcConnectionOptions build() {
- return new JdbcConnectionOptions(url, driverName, username, password);
+ return new JdbcConnectionOptions(url, driverName, username, password, connectionCheckTimeoutSeconds);
}
}
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
index 2cb5f1d..2823b6d 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
@@ -49,7 +49,6 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
-import static org.apache.flink.connector.jdbc.internal.options.JdbcOptions.CONNECTION_CHECK_TIMEOUT_SECONDS;
import static org.apache.flink.connector.jdbc.utils.JdbcUtils.setRecordToStatement;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -179,7 +178,7 @@ public class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStat
throw new IOException(e);
}
try {
- if (!connection.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) {
+ if (!connectionProvider.isConnectionValid()){
connection = connectionProvider.reestablishConnection();
jdbcStatementExecutor.closeStatements();
jdbcStatementExecutor.prepareStatements(connection);
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java
index f5a5e70..70f0d79 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java
@@ -29,4 +29,6 @@ public interface JdbcConnectionProvider {
Connection getConnection() throws Exception;
Connection reestablishConnection() throws Exception;
+
+ boolean isConnectionValid() throws Exception;
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java
index 4424054..c5bbd5f 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java
@@ -85,4 +85,9 @@ public class SimpleJdbcConnectionProvider implements JdbcConnectionProvider, Ser
connection = getConnection();
return connection;
}
+
+ @Override
+ public boolean isConnectionValid() throws Exception {
+ return connection.isValid(jdbcOptions.getConnectionCheckTimeoutSeconds());
+ }
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
index eeb4a0a..7299794 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
@@ -36,15 +36,12 @@ public class JdbcOptions extends JdbcConnectionOptions {
private static final long serialVersionUID = 1L;
- public static final int CONNECTION_CHECK_TIMEOUT_SECONDS = 60;
-
private String tableName;
private JdbcDialect dialect;
private final @Nullable Integer parallelism;
- private JdbcOptions(String dbURL, String tableName, String driverName, String username,
- String password, JdbcDialect dialect, Integer parallelism) {
- super(dbURL, driverName, username, password);
+ private JdbcOptions(String dbURL, String tableName, String driverName, String username, String password, JdbcDialect dialect, Integer parallelism, int connectionCheckTimeoutSeconds) {
+ super(dbURL, driverName, username, password, connectionCheckTimeoutSeconds);
this.tableName = tableName;
this.dialect = dialect;
this.parallelism = parallelism;
@@ -76,15 +73,17 @@ public class JdbcOptions extends JdbcConnectionOptions {
Objects.equals(username, options.username) &&
Objects.equals(password, options.password) &&
Objects.equals(dialect.getClass().getName(), options.dialect.getClass().getName()) &&
- Objects.equals(parallelism, options.parallelism);
+ Objects.equals(parallelism, options.parallelism) &&
+ Objects.equals(connectionCheckTimeoutSeconds, options.connectionCheckTimeoutSeconds);
} else {
return false;
}
+
}
@Override
public int hashCode() {
- return Objects.hash(url, tableName, driverName, username, password, dialect, parallelism);
+ return Objects.hash(url, tableName, driverName, username, password, dialect.getClass().getName(), parallelism, connectionCheckTimeoutSeconds);
}
/**
@@ -98,6 +97,7 @@ public class JdbcOptions extends JdbcConnectionOptions {
private String password;
private JdbcDialect dialect;
private Integer parallelism;
+ private int connectionCheckTimeoutSeconds = 60;
/**
* required, table name.
@@ -124,6 +124,14 @@ public class JdbcOptions extends JdbcConnectionOptions {
}
/**
+ * optional, connectionCheckTimeoutSeconds.
+ */
+ public Builder setConnectionCheckTimeoutSeconds(int connectionCheckTimeoutSeconds) {
+ this.connectionCheckTimeoutSeconds = connectionCheckTimeoutSeconds;
+ return this;
+ }
+
+ /**
* optional, driver name, dialect has a default driver name,
* See {@link JdbcDialect#defaultDriverName}.
*/
@@ -170,7 +178,7 @@ public class JdbcOptions extends JdbcConnectionOptions {
});
}
- return new JdbcOptions(dbURL, tableName, driverName, username, password, dialect, parallelism);
+ return new JdbcOptions(dbURL, tableName, driverName, username, password, dialect, parallelism, connectionCheckTimeoutSeconds);
}
}
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
index fa7190f..775cddc 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
@@ -80,6 +80,11 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam
.noDefaultValue()
.withDescription("the class name of the JDBC driver to use to connect to this URL. " +
"If not set, it will automatically be derived from the URL.");
+ public static final ConfigOption<Duration> MAX_RETRY_TIMEOUT = ConfigOptions
+ .key("connection.max-retry-timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(60))
+ .withDescription("Maximum timeout between retries.");
// read config options
private static final ConfigOption<String> SCAN_PARTITION_COLUMN = ConfigOptions
@@ -192,7 +197,8 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam
.setDBUrl(url)
.setTableName(readableConfig.get(TABLE_NAME))
.setDialect(JdbcDialects.get(url).get())
- .setParallelism(readableConfig.getOptional(FactoryUtil.SINK_PARALLELISM).orElse(null));
+ .setParallelism(readableConfig.getOptional(FactoryUtil.SINK_PARALLELISM).orElse(null))
+ .setConnectionCheckTimeoutSeconds((int) readableConfig.get(MAX_RETRY_TIMEOUT).getSeconds());
readableConfig.getOptional(DRIVER).ifPresent(builder::setDriverName);
readableConfig.getOptional(USERNAME).ifPresent(builder::setUsername);
@@ -274,6 +280,7 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam
optionalOptions.add(SINK_BUFFER_FLUSH_INTERVAL);
optionalOptions.add(SINK_MAX_RETRIES);
optionalOptions.add(FactoryUtil.SINK_PARALLELISM);
+ optionalOptions.add(MAX_RETRY_TIMEOUT);
return optionalOptions;
}
@@ -324,6 +331,13 @@ public class JdbcDynamicTableFactory implements DynamicTableSourceFactory, Dynam
SINK_MAX_RETRIES.key(),
config.get(SINK_MAX_RETRIES)));
}
+
+ if (config.get(MAX_RETRY_TIMEOUT).getSeconds() <= 0) {
+ throw new IllegalArgumentException(String.format(
+ "The value of '%s' option must be in second granularity and shouldn't be smaller than 1 second, but is %s.",
+ MAX_RETRY_TIMEOUT.key(),
+ config.get(ConfigOptions.key(MAX_RETRY_TIMEOUT.key()).stringType().noDefaultValue())));
+ }
}
private void checkAllOrNone(ReadableConfig config, ConfigOption<?>[] configOptions) {
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java
index 4946949..ff0dab0 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java
@@ -48,7 +48,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import static org.apache.flink.connector.jdbc.internal.options.JdbcOptions.CONNECTION_CHECK_TIMEOUT_SECONDS;
import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getFieldFromResultSet;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -74,6 +73,7 @@ public class JdbcLookupFunction extends TableFunction<Row> {
private final String dbURL;
private final String username;
private final String password;
+ private final int connectionCheckTimeoutSeconds;
private final TypeInformation[] keyTypes;
private final int[] keySqlTypes;
private final String[] fieldNames;
@@ -95,6 +95,7 @@ public class JdbcLookupFunction extends TableFunction<Row> {
this.dbURL = options.getDbURL();
this.username = options.getUsername().orElse(null);
this.password = options.getPassword().orElse(null);
+ this.connectionCheckTimeoutSeconds = options.getConnectionCheckTimeoutSeconds();
this.fieldNames = fieldNames;
this.fieldTypes = fieldTypes;
this.keyNames = keyNames;
@@ -177,7 +178,7 @@ public class JdbcLookupFunction extends TableFunction<Row> {
}
try {
- if (!dbConn.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) {
+ if (!dbConn.isValid(connectionCheckTimeoutSeconds)) {
statement.close();
dbConn.close();
establishConnectionAndStatement();
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java
index d00bb47..f192ed0 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java
@@ -50,7 +50,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import static org.apache.flink.connector.jdbc.internal.options.JdbcOptions.CONNECTION_CHECK_TIMEOUT_SECONDS;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -68,6 +67,7 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
private final String dbURL;
private final String username;
private final String password;
+ private final int connectionCheckTimeoutSeconds;
private final DataType[] keyTypes;
private final String[] keyNames;
private final long cacheMaxSize;
@@ -97,6 +97,7 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
this.username = options.getUsername().orElse(null);
this.password = options.getPassword().orElse(null);
this.keyNames = keyNames;
+ this.connectionCheckTimeoutSeconds = options.getConnectionCheckTimeoutSeconds();
List<String> nameList = Arrays.asList(fieldNames);
this.keyTypes = Arrays.stream(keyNames)
.map(s -> {
@@ -175,7 +176,7 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
}
try {
- if (!dbConn.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) {
+ if (!dbConn.isValid(connectionCheckTimeoutSeconds)) {
statement.close();
dbConn.close();
establishConnectionAndStatement();
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactory.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactory.java
index c74d0d9..b56b3c5 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactory.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactory.java
@@ -47,6 +47,7 @@ import static org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK;
import static org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK_ROWTIME;
import static org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK_STRATEGY_DATA_TYPE;
import static org.apache.flink.table.descriptors.DescriptorProperties.WATERMARK_STRATEGY_EXPR;
+import static org.apache.flink.table.descriptors.JdbcValidator.CONNECTOR_CONNECTION_MAX_RETRY_TIMEOUT;
import static org.apache.flink.table.descriptors.JdbcValidator.CONNECTOR_DRIVER;
import static org.apache.flink.table.descriptors.JdbcValidator.CONNECTOR_LOOKUP_CACHE_MAX_ROWS;
import static org.apache.flink.table.descriptors.JdbcValidator.CONNECTOR_LOOKUP_CACHE_TTL;
@@ -95,6 +96,7 @@ public class JdbcTableSourceSinkFactory implements
properties.add(CONNECTOR_TABLE);
properties.add(CONNECTOR_USERNAME);
properties.add(CONNECTOR_PASSWORD);
+ properties.add(CONNECTOR_CONNECTION_MAX_RETRY_TIMEOUT);
// scan options
properties.add(CONNECTOR_READ_QUERY);
@@ -182,6 +184,8 @@ public class JdbcTableSourceSinkFactory implements
.setTableName(descriptorProperties.getString(CONNECTOR_TABLE))
.setDialect(JdbcDialects.get(url).get());
+ descriptorProperties.getOptionalDuration(CONNECTOR_CONNECTION_MAX_RETRY_TIMEOUT).ifPresent(
+ s -> builder.setConnectionCheckTimeoutSeconds((int) s.getSeconds()));
descriptorProperties.getOptionalString(CONNECTOR_DRIVER).ifPresent(builder::setDriverName);
descriptorProperties.getOptionalString(CONNECTOR_USERNAME).ifPresent(builder::setUsername);
descriptorProperties.getOptionalString(CONNECTOR_PASSWORD).ifPresent(builder::setPassword);
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/table/descriptors/JdbcValidator.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/table/descriptors/JdbcValidator.java
index e80a5a7..827df14 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/table/descriptors/JdbcValidator.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/table/descriptors/JdbcValidator.java
@@ -42,6 +42,7 @@ public class JdbcValidator extends ConnectorDescriptorValidator {
public static final String CONNECTOR_DRIVER = "connector.driver";
public static final String CONNECTOR_USERNAME = "connector.username";
public static final String CONNECTOR_PASSWORD = "connector.password";
+ public static final String CONNECTOR_CONNECTION_MAX_RETRY_TIMEOUT = "connector.connection.max-retry-timeout";
public static final String CONNECTOR_READ_QUERY = "connector.read.query";
public static final String CONNECTOR_READ_PARTITION_COLUMN = "connector.read.partition.column";
@@ -73,6 +74,7 @@ public class JdbcValidator extends ConnectorDescriptorValidator {
properties.validateString(CONNECTOR_DRIVER, true);
properties.validateString(CONNECTOR_USERNAME, true);
properties.validateString(CONNECTOR_PASSWORD, true);
+ properties.validateDuration(CONNECTOR_CONNECTION_MAX_RETRY_TIMEOUT, true, 1000);
final String url = properties.getString(CONNECTOR_URL);
final Optional<JdbcDialect> dialect = JdbcDialects.get(url);
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java
index 63b2f3f..e97d9dc 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactoryTest.java
@@ -63,6 +63,7 @@ public class JdbcDynamicTableFactoryTest {
properties.put("driver", "org.apache.derby.jdbc.EmbeddedDriver");
properties.put("username", "user");
properties.put("password", "pass");
+ properties.put("connection.max-retry-timeout", "120s");
// validation for source
DynamicTableSource actualSource = createTableSource(properties);
@@ -72,6 +73,7 @@ public class JdbcDynamicTableFactoryTest {
.setDriverName("org.apache.derby.jdbc.EmbeddedDriver")
.setUsername("user")
.setPassword("pass")
+ .setConnectionCheckTimeoutSeconds(120)
.build();
JdbcLookupOptions lookupOptions = JdbcLookupOptions.builder()
.setCacheMaxSize(-1)
@@ -339,6 +341,18 @@ public class JdbcDynamicTableFactoryTest {
"The value of 'sink.max-retries' option shouldn't be negative, but is -1.")
.isPresent());
}
+
+ // connection.max-retry-timeout shouldn't be smaller than 1 second
+ try {
+ Map<String, String> properties = getAllOptions();
+ properties.put("connection.max-retry-timeout", "100ms");
+ createTableSource(properties);
+ fail("exception expected");
+ } catch (Throwable t) {
+ assertTrue(ExceptionUtils.findThrowableWithMessage(t,
+ "The value of 'connection.max-retry-timeout' option must be in second granularity and shouldn't be smaller than 1 second, but is 100ms.")
+ .isPresent());
+ }
}
private Map<String, String> getAllOptions() {
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactoryTest.java
index b4af2b1..51202e2 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcTableSourceSinkFactoryTest.java
@@ -65,6 +65,7 @@ public class JdbcTableSourceSinkFactoryTest {
properties.put("connector.driver", "org.apache.derby.jdbc.EmbeddedDriver");
properties.put("connector.username", "user");
properties.put("connector.password", "pass");
+ properties.put("connector.connection.max-retry-timeout", "120s");
final StreamTableSource<?> actual = TableFactoryService.find(StreamTableSourceFactory.class, properties)
.createStreamTableSource(properties);
@@ -75,6 +76,7 @@ public class JdbcTableSourceSinkFactoryTest {
.setDriverName("org.apache.derby.jdbc.EmbeddedDriver")
.setUsername("user")
.setPassword("pass")
+ .setConnectionCheckTimeoutSeconds(120)
.build();
final JdbcTableSource expected = JdbcTableSource.builder()
.setOptions(options)
@@ -267,6 +269,17 @@ public class JdbcTableSourceSinkFactoryTest {
fail("exception expected");
} catch (ValidationException ignored) {
}
+
+ // connection.max-retry-timeout property is smaller than 1 second
+ try {
+ Map<String, String> properties = getBasicProperties();
+ properties.put("connector.connection.max-retry-timeout", "100ms");
+
+ TableFactoryService.find(StreamTableSourceFactory.class, properties)
+ .createStreamTableSource(properties);
+ fail("exception expected");
+ } catch (ValidationException ignored) {
+ }
}
private Map<String, String> getBasicProperties() {