Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

🎉 Enhanced error messaging for DB source connectors #18087

Merged
merged 14 commits into from
Oct 19, 2022
Merged

🎉 Enhanced error messaging for DB source connectors #18087

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
ffb191e
Enhanced error messaging for invalid cursor state
akashkulk Oct 17, 2022
7f09224
Merge branch 'master' into akash/error
akashkulk Oct 17, 2022
554c99c
Handle SSH config fail
akashkulk Oct 18, 2022
6006d5d
Merge branch 'master' into akash/error
akashkulk Oct 18, 2022
98192f9
Handle SSH config fail
akashkulk Oct 18, 2022
6c47a5f
Merge branch 'master' into akash/error
akashkulk Oct 18, 2022
d8e9385
Update AbstractDbSource.java
akashkulk Oct 19, 2022
be9a9d1
Merge branch 'master' into akash/error
akashkulk Oct 19, 2022
f985d65
Bump up versions + doc for beta+ connectors
akashkulk Oct 19, 2022
195f0b2
Merge branch 'master' into akash/error
akashkulk Oct 19, 2022
fd4df08
auto-bump connector version
octavia-squidington-iii Oct 19, 2022
a8492eb
auto-bump connector version
octavia-squidington-iii Oct 19, 2022
090a659
Merge branch 'akash/error' of https://github.com/airbytehq/airbyte in…
octavia-squidington-iii Oct 19, 2022
ff3638a
auto-bump connector version
octavia-squidington-iii Oct 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions airbyte-config/init/src/main/resources/seed/source_definitions.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -637,7 +637,7 @@
- name: Microsoft SQL Server (MSSQL)
sourceDefinitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1
dockerRepository: airbyte/source-mssql
dockerImageTag: 0.4.21
dockerImageTag: 0.4.22
documentationUrl: https://docs.airbyte.com/integrations/sources/mssql
icon: mssql.svg
sourceType: database
Expand Down Expand Up @@ -685,7 +685,7 @@
- name: MySQL
sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
dockerRepository: airbyte/source-mysql
dockerImageTag: 1.0.5
dockerImageTag: 1.0.6
documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
icon: mysql.svg
sourceType: database
Expand Down Expand Up @@ -843,7 +843,7 @@
- name: Postgres
sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
dockerRepository: airbyte/source-postgres
dockerImageTag: 1.0.17
dockerImageTag: 1.0.18
documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
icon: postgresql.svg
sourceType: database
Expand Down
6 changes: 3 additions & 3 deletions airbyte-config/init/src/main/resources/seed/source_specs.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6152,7 +6152,7 @@
supportsNormalization: false
supportsDBT: false
supported_destination_sync_modes: []
- dockerImage: "airbyte/source-mssql:0.4.21"
- dockerImage: "airbyte/source-mssql:0.4.22"
spec:
documentationUrl: "https://docs.airbyte.com/integrations/destinations/mssql"
connectionSpecification:
Expand Down Expand Up @@ -6982,7 +6982,7 @@
supportsNormalization: false
supportsDBT: false
supported_destination_sync_modes: []
- dockerImage: "airbyte/source-mysql:1.0.5"
- dockerImage: "airbyte/source-mysql:1.0.6"
spec:
documentationUrl: "https://docs.airbyte.com/integrations/sources/mysql"
connectionSpecification:
Expand Down Expand Up @@ -8614,7 +8614,7 @@
supportsNormalization: false
supportsDBT: false
supported_destination_sync_modes: []
- dockerImage: "airbyte/source-postgres:1.0.17"
- dockerImage: "airbyte/source-postgres:1.0.18"
spec:
documentationUrl: "https://docs.airbyte.com/integrations/sources/postgres"
connectionSpecification:
Expand Down
13 changes: 12 additions & 1 deletion ...ions/bases/base-java/src/main/java/io/airbyte/integrations/base/ssh/SshWrappedSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,16 @@
import com.fasterxml.jackson.databind.JsonNode;
import io.airbyte.commons.util.AutoCloseableIterator;
import io.airbyte.commons.util.AutoCloseableIterators;
import io.airbyte.integrations.base.AirbyteTraceMessageUtility;
import io.airbyte.integrations.base.Source;
import io.airbyte.protocol.models.AirbyteCatalog;
import io.airbyte.protocol.models.AirbyteConnectionStatus;
import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
import io.airbyte.protocol.models.AirbyteMessage;
import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
import io.airbyte.protocol.models.ConnectorSpecification;
import java.util.List;
import org.apache.sshd.common.SshException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -37,7 +40,15 @@ public ConnectorSpecification spec() throws Exception {

@Override
public AirbyteConnectionStatus check(final JsonNode config) throws Exception {
return SshTunnel.sshWrap(config, hostKey, portKey, delegate::check);
try {
return SshTunnel.sshWrap(config, hostKey, portKey, delegate::check);
} catch (final SshException e) {
final String sshErrorMessage = "Could not connect with provided SSH configuration. Error: " + e.getMessage();
AirbyteTraceMessageUtility.emitConfigErrorTrace(e, sshErrorMessage);
return new AirbyteConnectionStatus()
.withStatus(Status.FAILED)
.withMessage(sshErrorMessage);
}
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mssql-strict-encrypt/Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,5 +16,5 @@ ENV APPLICATION source-mssql-strict-encrypt

COPY --from=build /airbyte /airbyte

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

COPY --from=build /airbyte /airbyte

LABEL io.airbyte.version=0.4.21
LABEL io.airbyte.version=0.4.22
LABEL io.airbyte.name=airbyte/source-mssql
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mysql-strict-encrypt/Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,6 @@ ENV APPLICATION source-mysql-strict-encrypt

COPY --from=build /airbyte /airbyte

LABEL io.airbyte.version=1.0.5
LABEL io.airbyte.version=1.0.6

LABEL io.airbyte.name=airbyte/source-mysql-strict-encrypt
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-mysql/Dockerfile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,6 @@ ENV APPLICATION source-mysql

COPY --from=build /airbyte /airbyte

LABEL io.airbyte.version=1.0.5
LABEL io.airbyte.version=1.0.6

LABEL io.airbyte.name=airbyte/source-mysql
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-postgres-strict-encrypt/Dockerfile
View file
Open in desktop
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=1.0.17
LABEL io.airbyte.version=1.0.18
LABEL io.airbyte.name=airbyte/source-postgres-strict-encrypt
2 changes: 1 addition & 1 deletion airbyte-integrations/connectors/source-postgres/Dockerfile
View file
Open in desktop
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=1.0.17
LABEL io.airbyte.version=1.0.18
LABEL io.airbyte.name=airbyte/source-postgres
71 changes: 42 additions & 29 deletions ...tional-db/src/main/java/io/airbyte/integrations/source/relationaldb/AbstractDbSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,36 +135,49 @@ public AirbyteCatalog discover(final JsonNode config) throws Exception {
public AutoCloseableIterator<AirbyteMessage> read(final JsonNode config,
final ConfiguredAirbyteCatalog catalog,
final JsonNode state)
throws Exception {
final StateManager stateManager =
StateManagerFactory.createStateManager(getSupportedStateType(config), deserializeInitialState(state, config), catalog);
final Instant emittedAt = Instant.now();

final Database database = createDatabaseInternal(config);

final Map<String, TableInfo<CommonField<DataType>>> fullyQualifiedTableNameToInfo =
discoverWithoutSystemTables(database)
.stream()
.collect(Collectors.toMap(t -> String.format("%s.%s", t.getNameSpace(), t.getName()), Function
.identity()));

validateCursorFieldForIncrementalTables(fullyQualifiedTableNameToInfo, catalog);

final List<AutoCloseableIterator<AirbyteMessage>> incrementalIterators =
getIncrementalIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager, emittedAt);
final List<AutoCloseableIterator<AirbyteMessage>> fullRefreshIterators =
getFullRefreshIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager, emittedAt);
final List<AutoCloseableIterator<AirbyteMessage>> iteratorList = Stream
.of(incrementalIterators, fullRefreshIterators)
.flatMap(Collection::stream)
.collect(Collectors.toList());
throws Exception {
try {
final StateManager stateManager =
StateManagerFactory.createStateManager(getSupportedStateType(config), deserializeInitialState(state, config), catalog);
final Instant emittedAt = Instant.now();

return AutoCloseableIterators
.appendOnClose(AutoCloseableIterators.concatWithEagerClose(iteratorList), () -> {
LOGGER.info("Closing database connection pool.");
Exceptions.toRuntime(this::close);
LOGGER.info("Closed database connection pool.");
});
final Database database = createDatabaseInternal(config);

final Map<String, TableInfo<CommonField<DataType>>> fullyQualifiedTableNameToInfo =
discoverWithoutSystemTables(database)
.stream()
.collect(Collectors.toMap(t -> String.format("%s.%s", t.getNameSpace(), t.getName()), Function
.identity()));

validateCursorFieldForIncrementalTables(fullyQualifiedTableNameToInfo, catalog);

final List<AutoCloseableIterator<AirbyteMessage>> incrementalIterators =
getIncrementalIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager, emittedAt);
final List<AutoCloseableIterator<AirbyteMessage>> fullRefreshIterators =
getFullRefreshIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager, emittedAt);
final List<AutoCloseableIterator<AirbyteMessage>> iteratorList = Stream
.of(incrementalIterators, fullRefreshIterators)
.flatMap(Collection::stream)
.collect(Collectors.toList());

return AutoCloseableIterators
.appendOnClose(AutoCloseableIterators.concatWithEagerClose(iteratorList), () -> {
LOGGER.info("Closing database connection pool.");
Exceptions.toRuntime(this::close);
LOGGER.info("Closed database connection pool.");
});
} catch (final Exception exception) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this logic of surfacing an AirbyteTraceMessage something that can be tested within PostgresSourceTest.java, MysqlSourceTest.java, etc? It seems there's already a test within PostgresSourceTest.java`

if (isConfigError(exception)) {
AirbyteTraceMessageUtility.emitConfigErrorTrace(exception, exception.getMessage());
}
throw exception;
akashkulk marked this conversation as resolved.
Show resolved Hide resolved
}
}

private boolean isConfigError(final Exception exception) {
// For now, enhanced error details should only be shown for InvalidCursorException. In the future, enhanced error messages will exist for
// additional error types.
return exception instanceof InvalidCursorException;
}

private void validateCursorFieldForIncrementalTables(final Map<String, TableInfo<CommonField<DataType>>> tableNameToTable,
Expand Down
Loading