Mysql source: handle ssh timeout exception

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -1078,7 +1078,7 @@
 - name: MySQL
   sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad
   dockerRepository: airbyte/source-mysql
-  dockerImageTag: 1.0.19
+  dockerImageTag: 1.0.20
   documentationUrl: https://docs.airbyte.com/integrations/sources/mysql
   icon: mysql.svg
   sourceType: database

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -9002,7 +9002,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mysql:1.0.19"
+- dockerImage: "airbyte/source-mysql:1.0.20"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/mysql"
     connectionSpecification:

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/ssh/SshTunnel.java

@@ -22,9 +22,11 @@
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import org.apache.sshd.client.SshClient;
 import org.apache.sshd.client.keyverifier.AcceptAllServerKeyVerifier;
 import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.SshException;
 import org.apache.sshd.common.util.net.SshdSocketAddress;
 import org.apache.sshd.common.util.security.SecurityUtils;
 import org.apache.sshd.core.CoreModuleProperties;
@@ -41,6 +43,8 @@
 public class SshTunnel implements AutoCloseable {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SshTunnel.class);
+  public static final String SSH_TIMEOUT_DISPLAY_MESSAGE =
+      "Timed out while opening a SSH Tunnel. Please double check the given SSH configurations and try again.";
 
   public enum TunnelMethod {
     NO_TUNNEL,
@@ -364,7 +368,13 @@ ClientSession openTunnel(final SshClient client) {
           remoteServiceHost, remoteServicePort, address.toInetSocketAddress()));
       return session;
     } catch (final IOException | GeneralSecurityException e) {
-      throw new RuntimeException(e);
+      if (e instanceof SshException && e.getMessage()
+          .toLowerCase(Locale.ROOT)
+          .contains("failed to get operation result within specified timeout")) {
+        throw new ConfigErrorException(SSH_TIMEOUT_DISPLAY_MESSAGE, e);
+      } else {
+        throw new RuntimeException(e);
+      }
     }
   }
 

airbyte-integrations/connectors/source-mysql-strict-encrypt/Dockerfile

@@ -16,6 +16,6 @@ ENV APPLICATION source-mysql-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=1.0.19
+LABEL io.airbyte.version=1.0.20
 
 LABEL io.airbyte.name=airbyte/source-mysql-strict-encrypt

airbyte-integrations/connectors/source-mysql/Dockerfile

@@ -16,6 +16,6 @@ ENV APPLICATION source-mysql
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=1.0.19
+LABEL io.airbyte.version=1.0.20
 
 LABEL io.airbyte.name=airbyte/source-mysql

airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/AbstractSshMySqlSourceAcceptanceTest.java

@@ -28,7 +28,7 @@ public abstract class AbstractSshMySqlSourceAcceptanceTest extends SourceAccepta
   private static final String STREAM_NAME = "id_and_name";
   private static final String STREAM_NAME2 = "starships";
 
-  private JsonNode config;
+  protected static JsonNode config;
 
   public abstract Path getConfigFilePath();
 

airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/SshPasswordMySqlSourceAcceptanceTest.java

@@ -4,10 +4,22 @@
 
 package io.airbyte.integrations.io.airbyte.integration_tests.sources;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.features.EnvVariableFeatureFlags;
+import io.airbyte.integrations.base.Source;
+import io.airbyte.integrations.base.ssh.SshBastionContainer;
+import io.airbyte.integrations.base.ssh.SshTunnel;
+import io.airbyte.integrations.source.mysql.MySqlSource;
 import io.airbyte.integrations.standardtest.source.TestDestinationEnv;
 import java.nio.file.Path;
+import java.util.List;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.containers.MySQLContainer;
+import org.testcontainers.containers.Network;
 import uk.org.webcompere.systemstubs.environment.EnvironmentVariables;
 import uk.org.webcompere.systemstubs.jupiter.SystemStub;
 import uk.org.webcompere.systemstubs.jupiter.SystemStubsExtension;
@@ -29,4 +41,32 @@ public Path getConfigFilePath() {
     return Path.of("secrets/ssh-pwd-repl-config.json");
   }
 
+  @Test
+  public void sshTimeoutExceptionMarkAsConfigErrorTest() throws Exception {
+    SshBastionContainer bastion = new SshBastionContainer();
+    final Network network = Network.newNetwork();
+    // set up env
+    MySQLContainer<?> db = startTestContainers(bastion, network);
+    config = bastion.getTunnelConfig(SshTunnel.TunnelMethod.SSH_PASSWORD_AUTH, bastion.getBasicDbConfigBuider(db, List.of("public")));
+    bastion.stopAndClose();
+    Source sshWrappedSource = MySqlSource.sshWrappedSource();
+    Exception exception = assertThrows(ConfigErrorException.class, () -> sshWrappedSource.discover(config));
+
+    String expectedMessage = "Timed out while opening a SSH Tunnel. Please double check the given SSH configurations and try again.";
+    String actualMessage = exception.getMessage();
+
+    assertTrue(actualMessage.contains(expectedMessage));
+  }
+
+  private MySQLContainer startTestContainers(SshBastionContainer bastion, Network network) {
+    bastion.initAndStartBastion(network);
+    return initAndStartJdbcContainer(network);
+  }
+
+  private MySQLContainer initAndStartJdbcContainer(Network network) {
+    MySQLContainer<?> db = new MySQLContainer<>("mysql:8.0").withNetwork(network);
+    db.start();
+    return db;
+  }
+
 }

docs/integrations/sources/mysql.md

@@ -256,8 +256,9 @@ WHERE actor_definition_id ='435bb9a5-7887-4809-aa58-28c27df0d7ad' AND (configura
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                                   |
 |:--------|:-----------|:-----------------------------------------------------------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| 1.0.19  | 2022-12-14 | [20436](https://github.com/airbytehq/airbyte/pull/20346)   | Consolidate date/time values mapping for JDBC sources                          |
-| 1.0.18  | 2022-12-14 | [20378](https://github.com/airbytehq/airbyte/pull/20378)   | Improve descriptions                                               |
+| 1.0.20  | 2023-01-24 | [20593](https://github.com/airbytehq/airbyte/pull/20593)   | Handle ssh time out exception                                                                                                                                             |
+| 1.0.19  | 2022-12-14 | [20436](https://github.com/airbytehq/airbyte/pull/20346)   | Consolidate date/time values mapping for JDBC sources                                                                                                                     |
+| 1.0.18  | 2022-12-14 | [20378](https://github.com/airbytehq/airbyte/pull/20378)   | Improve descriptions                                                                                                                                                      |
 | 1.0.17  | 2022-12-13 | [20289](https://github.com/airbytehq/airbyte/pull/20289)   | Mark unknown column exception as config error                                                                                                                             |
 | 1.0.16  | 2022-12-12 | [18959](https://github.com/airbytehq/airbyte/pull/18959)   | CDC : Don't timeout if snapshot is not complete.                                                                                                                          |
 | 1.0.15  | 2022-12-06 | [20000](https://github.com/airbytehq/airbyte/pull/20000)   | Add check and better messaging when user does not have permission to access binary log in CDC mode                                                                        |