wuchong commented on a change in pull request #14387:
URL: https://github.com/apache/flink/pull/14387#discussion_r545778061
##########
File path: docs/dev/table/connectors/jdbc.md
##########
@@ -148,6 +148,13 @@ Connector Options
<td>String</td>
<td>The JDBC password.</td>
</tr>
+ <tr>
+ <td><h5>max-retry-timeout</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">60s</td>
+ <td>Duration</td>
+ <td>The JDBC connectionCheckTimeoutSeconds.</td>
Review comment:
`Maximum timeout between retries.`
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
##########
@@ -179,11 +178,9 @@ public synchronized void flush() throws IOException {
throw new IOException(e);
}
try {
- if
(!connection.isValid(connectionCheckTimeoutSeconds)) {
Review comment:
Why remove this?
##########
File path: docs/dev/table/connectors/jdbc.md
##########
@@ -148,6 +148,13 @@ Connector Options
<td>String</td>
<td>The JDBC password.</td>
</tr>
+ <tr>
+ <td><h5>max-retry-timeout</h5></td>
Review comment:
```suggestion
<td><h5>connection.max-retry-timeout</h5></td>
```
##########
File path: docs/dev/table/connectors/jdbc.zh.md
##########
@@ -148,6 +148,13 @@ Connector Options
<td>String</td>
<td>The JDBC password.</td>
</tr>
+ <tr>
+ <td><h5>max-retry-timeout</h5></td>
Review comment:
```suggestion
<td><h5>connection.max-retry-timeout</h5></td>
```
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -67,11 +68,22 @@ public boolean equals(Object o) {
Objects.equals(driverName, options.driverName)
&&
Objects.equals(username, options.username) &&
Objects.equals(password, options.password) &&
- Objects.equals(dialect.getClass().getName(),
options.dialect.getClass().getName());
+ Objects.equals(dialect.getClass().getName(),
options.dialect.getClass().getName()) &&
+ Objects.equals(connectionCheckTimeoutSeconds,
options.connectionCheckTimeoutSeconds);
} else {
return false;
}
}
+ @Override
Review comment:
Add empty line before this.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
##########
@@ -196,6 +196,7 @@ private JdbcOptions getJdbcOptions(ReadableConfig
readableConfig) {
final JdbcOptions.Builder builder = JdbcOptions.builder()
.setDBUrl(url)
.setTableName(readableConfig.get(TABLE_NAME))
+ .setConnectionCheckTimeoutSeconds((int)
readableConfig.get(MAX_RETRY_TIMEOUT).toMillis())
Review comment:
This is wrong. `toMillis()` returns milliseconds but the parameter
requires seconds. Should use `toSeconds()` here.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -53,6 +53,7 @@ public String getTableName() {
public JdbcDialect getDialect() {
return dialect;
}
+ public int getConnectionCheckTimeoutSeconds() { return
connectionCheckTimeoutSeconds; }
Review comment:
```suggestion
public int getConnectionCheckTimeoutSeconds() { return
connectionCheckTimeoutSeconds; }
```
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -107,7 +119,13 @@ public Builder setPassword(String password) {
this.password = password;
return this;
}
-
+ /**
Review comment:
Add emtpy line before method.
##########
File path: docs/dev/table/connectors/jdbc.zh.md
##########
@@ -148,6 +148,13 @@ Connector Options
<td>String</td>
<td>The JDBC password.</td>
</tr>
+ <tr>
+ <td><h5>max-retry-timeout</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">60s</td>
+ <td>Duration</td>
+ <td>The JDBC connectionCheckTimeoutSeconds.</td>
Review comment:
`Maximum timeout between retries.`
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -150,7 +168,7 @@ public JdbcOptions build() {
});
}
- return new JdbcOptions(dbURL, tableName, driverName,
username, password, dialect);
+ return new JdbcOptions(dbURL, tableName, driverName,
username, password, dialect,connectionCheckTimeoutSeconds);
Review comment:
Add space before `connectionCheckTimeoutSeconds`.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
##########
@@ -80,6 +80,11 @@
.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("max-retry-timeout")
Review comment:
Would be better to use `connection.max-retry-timeout` to make it clear
that this is a timeout config for connection. This also keeps align with
Elasticsearch `connection.max-retry-timeout` config.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -83,7 +95,7 @@ public boolean equals(Object o) {
private String username;
private String password;
private JdbcDialect dialect;
-
+ private int connectionCheckTimeoutSeconds;
Review comment:
Would be better to add a default 60 value for this variable. Otherwise,
if users are using DataStream API, they will get a zero timeout by default
which means never timeout.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -67,11 +68,22 @@ public boolean equals(Object o) {
Objects.equals(driverName, options.driverName)
&&
Objects.equals(username, options.username) &&
Objects.equals(password, options.password) &&
- Objects.equals(dialect.getClass().getName(),
options.dialect.getClass().getName());
+ Objects.equals(dialect.getClass().getName(),
options.dialect.getClass().getName()) &&
+ Objects.equals(connectionCheckTimeoutSeconds,
options.connectionCheckTimeoutSeconds);
} else {
return false;
}
}
+ @Override
+ public int hashCode() {
+ return Objects.hash(url,
+ tableName,
+ driverName,
+ username,
+ password,
+ dialect,
Review comment:
Use `dialect.getClass().getName()` instead of `dialect`, otherwise, the
`hashcode` is not compliant with `equals`.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
##########
@@ -53,6 +53,7 @@ public String getTableName() {
public JdbcDialect getDialect() {
return dialect;
}
+ public int getConnectionCheckTimeoutSeconds() { return
connectionCheckTimeoutSeconds; }
Review comment:
Add emtpy line before this method.
##########
File path:
flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableFactory.java
##########
@@ -196,6 +196,7 @@ private JdbcOptions getJdbcOptions(ReadableConfig
readableConfig) {
final JdbcOptions.Builder builder = JdbcOptions.builder()
.setDBUrl(url)
.setTableName(readableConfig.get(TABLE_NAME))
+ .setConnectionCheckTimeoutSeconds((int)
readableConfig.get(MAX_RETRY_TIMEOUT).toMillis())
Review comment:
We should better to check the configured timeout is greater than 1
second. Otherwise, we will get a zero seconds which means never timeout for
`Connection#isValid(timeout)`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
