Skip to content

Commit

Permalink
Make connectionTimeoutMs configurable (#15226)
Browse files Browse the repository at this point in the history 
* Extract connectionTimeout from jdbc_url_params along with corresponding tests

* Fixed linter issues

* Reverted createDataSourceWithConnectionTimeout and migrated logic to get operation

* Fixed dangling createDataSourceWithConnectionTimeout and linter issues

* Fixed import to use java standard library

* Bump Postgres Source and Postgres Source Strict Encrypt versions

* Fixed import ordering issues

* Bumped the connector version [CI fix] for definitions not generated
  • Loading branch information
@ryankfu
ryankfu authored Aug 4, 2022
1 parent 25871c2 commit 23bdd61
Show file tree
Hide file tree
Showing 7 changed files with 127 additions and 50 deletions.
2 changes: 1 addition & 1 deletion airbyte-config/init/src/main/resources/seed/source_definitions.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -762,7 +762,7 @@
- name: Postgres
sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
dockerRepository: airbyte/source-postgres
dockerImageTag: 0.4.42
dockerImageTag: 0.4.43
documentationUrl: https://docs.airbyte.io/integrations/sources/postgres
icon: postgresql.svg
sourceType: database
Expand Down
2 changes: 1 addition & 1 deletion airbyte-config/init/src/main/resources/seed/source_specs.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7140,7 +7140,7 @@
supportsNormalization: false
supportsDBT: false
supported_destination_sync_modes: []
- dockerImage: "airbyte/source-postgres:0.4.42"
- dockerImage: "airbyte/source-postgres:0.4.43"
spec:
documentationUrl: "https://docs.airbyte.com/integrations/sources/postgres"
connectionSpecification:
Expand Down
35 changes: 33 additions & 2 deletions airbyte-db/db-lib/src/main/java/io/airbyte/db/factory/DataSourceFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import java.io.Closeable;
import java.time.Duration;
import java.util.Map;
import javax.sql.DataSource;

Expand Down Expand Up @@ -60,6 +61,7 @@ public static DataSource create(final String username,
.withJdbcUrl(jdbcConnectionString)
.withPassword(password)
.withUsername(username)
.withConnectionTimeoutMs(DataSourceBuilder.getConnectionTimeoutMs(connectionProperties))
.build();
}

Expand Down Expand Up @@ -173,14 +175,43 @@ private static class DataSourceBuilder {
private String jdbcUrl;
private int maximumPoolSize = 10;
private int minimumPoolSize = 0;
// the default 30000 millisecond is sometimes not enough for the acceptance test
private long connectionTimeoutMs = 60000;
private long connectionTimeoutMs;
private String password;
private int port = 5432;
private String username;
private static final String CONNECT_TIMEOUT_KEY = "connectTimeout";
private static final Duration CONNECT_TIMEOUT_DEFAULT = Duration.ofSeconds(60);

private DataSourceBuilder() {}

/**
* Retrieves connectionTimeout value from connection properties in seconds, default minimum timeout
* is 60 seconds since Hikari default of 30 seconds is not enough for acceptance tests. In the case
* the value is 0, pass the value along as Hikari and Postgres use default max value for 0 timeout
* value
*
* NOTE: HikariCP uses milliseconds for all time values:
* https://github.com/brettwooldridge/HikariCP#gear-configuration-knobs-baby whereas Postgres is
* measured in seconds: https://jdbc.postgresql.org/documentation/head/connect.html
*
* @param connectionProperties custom jdbc_url_parameters containing information on connection
* properties
* @return DataSourceBuilder class used to create dynamic fields for DataSource
*/
private static long getConnectionTimeoutMs(final Map<String, String> connectionProperties) {
final Duration connectionTimeout;
// TODO: the usage of CONNECT_TIMEOUT_KEY is Postgres specific, may need to extend for other
// databases
connectionTimeout =
connectionProperties.containsKey(CONNECT_TIMEOUT_KEY) ? Duration.ofSeconds(Long.parseLong(connectionProperties.get(CONNECT_TIMEOUT_KEY)))
: CONNECT_TIMEOUT_DEFAULT;
if (connectionTimeout.getSeconds() == 0) {
return connectionTimeout.toMillis();
} else {
return (connectionTimeout.compareTo(CONNECT_TIMEOUT_DEFAULT) > 0 ? connectionTimeout : CONNECT_TIMEOUT_DEFAULT).toMillis();
}
}

public DataSourceBuilder withConnectionProperties(final Map<String, String> connectionProperties) {
if (connectionProperties != null) {
this.connectionProperties = connectionProperties;
Expand Down
112 changes: 79 additions & 33 deletions airbyte-db/db-lib/src/test/java/io/airbyte/db/factory/DataSourceFactoryTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,20 +15,94 @@
import java.util.Map;
import javax.sql.DataSource;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

/**
* Test suite for the {@link DataSourceFactory} class.
*/
class DataSourceFactoryTest extends CommonFactoryTest {

static String database;
static String driverClassName;
static String host;
static String jdbcUrl;
static String password;
static Integer port;
static String username;

@BeforeAll
static void setup() {
host = container.getHost();
port = container.getFirstMappedPort();
database = container.getDatabaseName();
username = container.getUsername();
password = container.getPassword();
driverClassName = container.getDriverClassName();
jdbcUrl = container.getJdbcUrl();
}

@Test
void testCreatingADataSourceWithJdbcUrl() {
final String username = container.getUsername();
final String password = container.getPassword();
final String driverClassName = container.getDriverClassName();
final String jdbcUrl = container.getJdbcUrl();
void testCreatingDataSourceWithConnectionTimeoutSetAboveDefault() {
final Map<String, String> connectionProperties = Map.of(
"connectTimeout", "61");
final DataSource dataSource = DataSourceFactory.create(
username,
password,
driverClassName,
jdbcUrl,
connectionProperties);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
assertEquals(61000, ((HikariDataSource) dataSource).getHikariConfigMXBean().getConnectionTimeout());
}

@Test
void testCreatingDataSourceWithConnectionTimeoutSetBelowDefault() {
final Map<String, String> connectionProperties = Map.of(
"connectTimeout", "30");
final DataSource dataSource = DataSourceFactory.create(
username,
password,
driverClassName,
jdbcUrl,
connectionProperties);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
assertEquals(60000, ((HikariDataSource) dataSource).getHikariConfigMXBean().getConnectionTimeout());
}

@Test
void testCreatingDataSourceWithConnectionTimeoutSetWithZero() {
final Map<String, String> connectionProperties = Map.of(
"connectTimeout", "0");
final DataSource dataSource = DataSourceFactory.create(
username,
password,
driverClassName,
jdbcUrl,
connectionProperties);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
assertEquals(Integer.MAX_VALUE, ((HikariDataSource) dataSource).getHikariConfigMXBean().getConnectionTimeout());
}

@Test
void testCreatingDataSourceWithConnectionTimeoutNotSet() {
final Map<String, String> connectionProperties = Map.of();
final DataSource dataSource = DataSourceFactory.create(
username,
password,
driverClassName,
jdbcUrl,
connectionProperties);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
assertEquals(60000, ((HikariDataSource) dataSource).getHikariConfigMXBean().getConnectionTimeout());
}

@Test
void testCreatingADataSourceWithJdbcUrl() {
final DataSource dataSource = DataSourceFactory.create(username, password, driverClassName, jdbcUrl);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
Expand All @@ -37,10 +111,6 @@ void testCreatingADataSourceWithJdbcUrl() {

@Test
void testCreatingADataSourceWithJdbcUrlAndConnectionProperties() {
final String username = container.getUsername();
final String password = container.getPassword();
final String driverClassName = container.getDriverClassName();
final String jdbcUrl = container.getJdbcUrl();
final Map<String, String> connectionProperties = Map.of("foo", "bar");

final DataSource dataSource = DataSourceFactory.create(username, password, driverClassName, jdbcUrl, connectionProperties);
Expand All @@ -51,13 +121,6 @@ void testCreatingADataSourceWithJdbcUrlAndConnectionProperties() {

@Test
void testCreatingADataSourceWithHostAndPort() {
final String username = container.getUsername();
final String password = container.getPassword();
final String driverClassName = container.getDriverClassName();
final String host = container.getHost();
final Integer port = container.getFirstMappedPort();
final String database = container.getDatabaseName();

final DataSource dataSource = DataSourceFactory.create(username, password, host, port, database, driverClassName);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
Expand All @@ -66,12 +129,6 @@ void testCreatingADataSourceWithHostAndPort() {

@Test
void testCreatingADataSourceWithHostPortAndConnectionProperties() {
final String username = container.getUsername();
final String password = container.getPassword();
final String driverClassName = container.getDriverClassName();
final String host = container.getHost();
final Integer port = container.getFirstMappedPort();
final String database = container.getDatabaseName();
final Map<String, String> connectionProperties = Map.of("foo", "bar");

final DataSource dataSource = DataSourceFactory.create(username, password, host, port, database, driverClassName, connectionProperties);
Expand All @@ -82,12 +139,7 @@ void testCreatingADataSourceWithHostPortAndConnectionProperties() {

@Test
void testCreatingAnInvalidDataSourceWithHostAndPort() {
final String username = container.getUsername();
final String password = container.getPassword();
final String driverClassName = "Unknown";
final String host = container.getHost();
final Integer port = container.getFirstMappedPort();
final String database = container.getDatabaseName();

assertThrows(RuntimeException.class, () -> {
DataSourceFactory.create(username, password, host, port, database, driverClassName);
Expand All @@ -96,12 +148,6 @@ void testCreatingAnInvalidDataSourceWithHostAndPort() {

@Test
void testCreatingAPostgresqlDataSource() {
final String username = container.getUsername();
final String password = container.getPassword();
final String host = container.getHost();
final Integer port = container.getFirstMappedPort();
final String database = container.getDatabaseName();

final DataSource dataSource = DataSourceFactory.createPostgres(username, password, host, port, database);
assertNotNull(dataSource);
assertEquals(HikariDataSource.class, dataSource.getClass());
Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-postgres-strict-encrypt/Dockerfile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,5 +16,5 @@ ENV APPLICATION source-postgres-strict-encrypt

COPY --from=build /airbyte /airbyte

LABEL io.airbyte.version=0.4.42
LABEL io.airbyte.version=0.4.43
LABEL io.airbyte.name=airbyte/source-postgres-strict-encrypt
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-postgres/Dockerfile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,5 +16,5 @@ ENV APPLICATION source-postgres

COPY --from=build /airbyte /airbyte

LABEL io.airbyte.version=0.4.42
LABEL io.airbyte.version=0.4.43
LABEL io.airbyte.name=airbyte/source-postgres
22 changes: 11 additions & 11 deletions docs/integrations/sources/postgres.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -353,17 +353,17 @@ Possible solutions include:
## Changelog

| Version | Date | Pull Request | Subject |
|:--------|:-----------|:---------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------|
| 0.4.42 | 2022-08-03 | [15273](https://github.com/airbytehq/airbyte/pull/15273) | Fix a bug in `0.4.36` and correctly parse the CDC initial record waiting time |
| 0.4.41 | 2022-08-03 | [15077](https://github.com/airbytehq/airbyte/pull/15077) | Sync data from beginning if the LSN is no longer valid in CDC |
| | 2022-08-03 | [14903](https://github.com/airbytehq/airbyte/pull/14903) | Emit state messages more frequently |
| 0.4.40 | 2022-08-03 | [15187](https://github.com/airbytehq/airbyte/pull/15187) | Add support for BCE dates/timestamps |
| | 2022-08-03 | [14534](https://github.com/airbytehq/airbyte/pull/14534) | Align regular and CDC integration tests and data mappers |
| 0.4.39 | 2022-08-02 | [14801](https://github.com/airbytehq/airbyte/pull/14801) | Fix multiply log bindings |
| 0.4.38 | 2022-07-26 | [14362](https://github.com/airbytehq/airbyte/pull/14362) | Integral columns are now discovered as int64 fields. |
| 0.4.37 | 2022-07-22 | [14714](https://github.com/airbytehq/airbyte/pull/14714) | Clarified error message when invalid cursor column selected |
| 0.4.36 | 2022-07-21 | [14451](https://github.com/airbytehq/airbyte/pull/14451) | Make initial CDC waiting time configurable (⛔ this version has a bug and will not work; use `0.4.42` instead) |
| 0.4.35 | 2022-07-14 | [14574](https://github.com/airbytehq/airbyte/pull/14574) | Removed additionalProperties:false from JDBC source connectors |
| 0.4.43 | 2022-08-03 | [15226](https://github.com/airbytehq/airbyte/pull/15226) | Make connectionTimeoutMs configurable through JDBC url parameters |
| 0.4.42 | 2022-08-03 | [15273](https://github.com/airbytehq/airbyte/pull/15273) | Fix a bug in `0.4.36` and correctly parse the CDC initial record waiting time |
| 0.4.41 | 2022-08-03 | [15077](https://github.com/airbytehq/airbyte/pull/15077) | Sync data from beginning if the LSN is no longer valid in CDC |
| | 2022-08-03 | [14903](https://github.com/airbytehq/airbyte/pull/14903) | Emit state messages more frequently |
| 0.4.40 | 2022-08-03 | [15187](https://github.com/airbytehq/airbyte/pull/15187) | Add support for BCE dates/timestamps |
| | 2022-08-03 | [14534](https://github.com/airbytehq/airbyte/pull/14534) | Align regular and CDC integration tests and data mappers |
| 0.4.39 | 2022-08-02 | [14801](https://github.com/airbytehq/airbyte/pull/14801) | Fix multiply log bindings |
| 0.4.38 | 2022-07-26 | [14362](https://github.com/airbytehq/airbyte/pull/14362) | Integral columns are now discovered as int64 fields. |
| 0.4.37 | 2022-07-22 | [14714](https://github.com/airbytehq/airbyte/pull/14714) | Clarified error message when invalid cursor column selected |
| 0.4.36 | 2022-07-21 | [14451](https://github.com/airbytehq/airbyte/pull/14451) | Make initial CDC waiting time configurable (⛔ this version has a bug and will not work; use `0.4.42` instead) |
| 0.4.35 | 2022-07-14 | [14574](https://github.com/airbytehq/airbyte/pull/14574) | Removed additionalProperties:false from JDBC source connectors |
| 0.4.34 | 2022-07-17 | [13840](https://github.com/airbytehq/airbyte/pull/13840) | Added the ability to connect using different SSL modes and SSL certificates. |
| 0.4.33 | 2022-07-14 | [14586](https://github.com/airbytehq/airbyte/pull/14586) | Validate source JDBC url parameters |
| 0.4.32 | 2022-07-07 | [14694](https://github.com/airbytehq/airbyte/pull/14694) | Force to produce LEGACY state if the use stream capable feature flag is set to false |
Expand Down

0 comments on commit 23bdd61

Please sign in to comment.