Set app_id in snowflake source connector based on oss/cloud (#19314)

* Set app_id in snowflake source connector based on oss/cloud Set app_id in snowflake source connector based on * Bump version + docs * auto-bump connector version * Formatting Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>

Author: akashkulk

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

@@ -1343,7 +1343,7 @@
 - name: Snowflake
  sourceDefinitionId: e2d65910-8c8b-40a1-ae7d-ee2416b2bfa2
  dockerRepository: airbyte/source-snowflake
- dockerImageTag: 0.1.25
+ dockerImageTag: 0.1.26
  documentationUrl: https://docs.airbyte.com/integrations/sources/snowflake
  icon: snowflake.svg
  sourceType: database

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

@@ -12728,7 +12728,7 @@
  - - "client_secret"
  oauthFlowOutputParameters:
  - - "refresh_token"
-- dockerImage: "airbyte/source-snowflake:0.1.25"
+- dockerImage: "airbyte/source-snowflake:0.1.26"
  spec:
  documentationUrl: "https://docs.airbyte.com/integrations/sources/snowflake"
  connectionSpecification:

airbyte-integrations/connectors/source-snowflake/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION source-snowflake
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.1.25
+LABEL io.airbyte.version=0.1.26
 LABEL io.airbyte.name=airbyte/source-snowflake

airbyte-integrations/connectors/source-snowflake/build.gradle

@@ -6,7 +6,7 @@ plugins {
 }
 
 application {
- mainClass = 'io.airbyte.integrations.source.snowflake.SnowflakeSource'
+ mainClass = 'io.airbyte.integrations.source.snowflake.SnowflakeSourceRunner'
  applicationDefaultJvmArgs = ['-XX:+ExitOnOutOfMemoryError', '-XX:MaxRAMPercentage=75.0']
 }
 

airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeDataSourceUtils.java

@@ -34,8 +34,10 @@ public class SnowflakeDataSourceUtils {
  public static final String OAUTH_METHOD = "OAuth";
  public static final String USERNAME_PASSWORD_METHOD = "username/password";
  public static final String UNRECOGNIZED = "Unrecognized";
+ public static final String AIRBYTE_OSS = "airbyte_oss";
+ public static final String AIRBYTE_CLOUD = "airbyte_cloud";
  private static final String JDBC_CONNECTION_STRING =
- "role=%s&warehouse=%s&database=%s&schema=%s&JDBC_QUERY_RESULT_FORMAT=%s&CLIENT_SESSION_KEEP_ALIVE=%s&application=Airbyte_Connector";
+ "role=%s&warehouse=%s&database=%s&schema=%s&JDBC_QUERY_RESULT_FORMAT=%s&CLIENT_SESSION_KEEP_ALIVE=%s&application=%s";
 
  private static final Logger LOGGER = LoggerFactory.getLogger(SnowflakeDataSourceUtils.class);
  private static final int PAUSE_BETWEEN_TOKEN_REFRESH_MIN = 7; // snowflake access token's TTL is 10min and can't be modified
@@ -53,9 +55,9 @@ public class SnowflakeDataSourceUtils {
  * @param config source config JSON
  * @return datasource
  */
- public static HikariDataSource createDataSource(final JsonNode config) {
+ public static HikariDataSource createDataSource(final JsonNode config, final String airbyteEnvironment) {
  final HikariDataSource dataSource = new HikariDataSource();
- dataSource.setJdbcUrl(buildJDBCUrl(config));
+ dataSource.setJdbcUrl(buildJDBCUrl(config, airbyteEnvironment));
 
  if (config.has("credentials")) {
  final JsonNode credentials = config.get("credentials");
@@ -130,7 +132,7 @@ public static String getAccessTokenUsingRefreshToken(final String hostName,
  }
  }
 
- public static String buildJDBCUrl(final JsonNode config) {
+ public static String buildJDBCUrl(final JsonNode config, final String airbyteEnvironment) {
  final StringBuilder jdbcUrl = new StringBuilder(String.format("jdbc:snowflake://%s/?",
  config.get(JdbcUtils.HOST_KEY).asText()));
 
@@ -143,7 +145,8 @@ public static String buildJDBCUrl(final JsonNode config) {
  // Needed for JDK17 - see
  // https://stackoverflow.com/questions/67409650/snowflake-jdbc-driver-internal-error-fail-to-retrieve-row-count-for-first-arrow
  "JSON",
- true));
+ true,
+ airbyteEnvironment));
 
  // https://docs.snowflake.com/en/user-guide/jdbc-configure.html#jdbc-driver-connection-string
  if (config.has(JdbcUtils.JDBC_URL_PARAMS_KEY)) {

airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeSource.java

@@ -16,7 +16,6 @@
 import io.airbyte.db.jdbc.JdbcUtils;
 import io.airbyte.db.jdbc.StreamingJdbcDatabase;
 import io.airbyte.db.jdbc.streaming.AdaptiveStreamingQueryConfig;
-import io.airbyte.integrations.base.IntegrationRunner;
 import io.airbyte.integrations.base.Source;
 import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
 import java.io.IOException;
@@ -37,16 +36,11 @@ public class SnowflakeSource extends AbstractJdbcSource<JDBCType> implements Sou
  public static final String DRIVER_CLASS = DatabaseDriver.SNOWFLAKE.getDriverClassName();
  public static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1);
 
- public SnowflakeSource() {
- super(DRIVER_CLASS, AdaptiveStreamingQueryConfig::new, new SnowflakeSourceOperations());
- }
+ private final String airbyteEnvironment;
 
- public static void main(final String[] args) throws Exception {
- final Source source = new SnowflakeSource();
- LOGGER.info("starting source: {}", SnowflakeSource.class);
- new IntegrationRunner(source).run(args);
- SCHEDULED_EXECUTOR_SERVICE.shutdownNow();
- LOGGER.info("completed source: {}", SnowflakeSource.class);
+ public SnowflakeSource(final String airbyteEnvironment) {
+ super(DRIVER_CLASS, AdaptiveStreamingQueryConfig::new, new SnowflakeSourceOperations());
+ this.airbyteEnvironment = airbyteEnvironment;
  }
 
  @Override
@@ -59,14 +53,14 @@ public JdbcDatabase createDatabase(final JsonNode config) throws SQLException {
 
  @Override
  protected DataSource createDataSource(final JsonNode config) {
- final DataSource dataSource = SnowflakeDataSourceUtils.createDataSource(config);
+ final DataSource dataSource = SnowflakeDataSourceUtils.createDataSource(config, airbyteEnvironment);
  dataSources.add(dataSource);
  return dataSource;
  }
 
  @Override
  public JsonNode toDatabaseConfig(final JsonNode config) {
- final String jdbcUrl = SnowflakeDataSourceUtils.buildJDBCUrl(config);
+ final String jdbcUrl = SnowflakeDataSourceUtils.buildJDBCUrl(config, airbyteEnvironment);
 
  if (config.has("credentials")) {
  final JsonNode credentials = config.get("credentials");

airbyte-integrations/connectors/source-snowflake/src/main/java/io.airbyte.integrations.source.snowflake/SnowflakeSourceRunner.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.source.snowflake;
+
+import static io.airbyte.integrations.source.snowflake.SnowflakeDataSourceUtils.AIRBYTE_CLOUD;
+import static io.airbyte.integrations.source.snowflake.SnowflakeDataSourceUtils.AIRBYTE_OSS;
+import static io.airbyte.integrations.source.snowflake.SnowflakeSource.SCHEDULED_EXECUTOR_SERVICE;
+
+import io.airbyte.integrations.base.adaptive.AdaptiveSourceRunner;
+
+public class SnowflakeSourceRunner {
+
+ public static void main(final String[] args) throws Exception {
+ AdaptiveSourceRunner.baseOnEnv()
+ .withOssSource(() -> new SnowflakeSource(AIRBYTE_OSS))
+ .withCloudSource(() -> new SnowflakeSource(AIRBYTE_CLOUD))
+ .run(args);
+ SCHEDULED_EXECUTOR_SERVICE.shutdownNow();
+ }
+
+}

airbyte-integrations/connectors/source-snowflake/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/SnowflakeJdbcSourceAcceptanceTest.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.io.airbyte.integration_tests.sources;
 
+import static io.airbyte.integrations.source.snowflake.SnowflakeDataSourceUtils.AIRBYTE_OSS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -102,7 +103,7 @@ public String getDriverClass() {
 
  @Override
  public AbstractJdbcSource<JDBCType> getJdbcSource() {
- return new SnowflakeSource();
+ return new SnowflakeSource(AIRBYTE_OSS);
  }
 
  @Test

airbyte-integrations/connectors/source-snowflake/src/test/java/io/airbyte/integrations/source/snowflake/SnowflakeDataSourceUtilsTest.java

@@ -30,24 +30,24 @@ class SnowflakeDataSourceUtilsTest {
  }
  """;
  private final String expectedJdbcUrl =
- "jdbc:snowflake://host/?role=role&warehouse=WAREHOUSE&database=DATABASE&schema=SOURCE_SCHEMA&JDBC_QUERY_RESULT_FORMAT=JSON&CLIENT_SESSION_KEEP_ALIVE=true&application=Airbyte_Connector";
+ "jdbc:snowflake://host/?role=role&warehouse=WAREHOUSE&database=DATABASE&schema=SOURCE_SCHEMA&JDBC_QUERY_RESULT_FORMAT=JSON&CLIENT_SESSION_KEEP_ALIVE=true&application=airbyte_oss";
 
  @Test
  void testBuildJDBCUrl() {
- JsonNode expectedConfig = Jsons.deserialize(config);
+ final JsonNode expectedConfig = Jsons.deserialize(config);
 
- String jdbcURL = SnowflakeDataSourceUtils.buildJDBCUrl(expectedConfig);
+ final String jdbcURL = SnowflakeDataSourceUtils.buildJDBCUrl(expectedConfig, SnowflakeDataSourceUtils.AIRBYTE_OSS);
 
  assertEquals(expectedJdbcUrl, jdbcURL);
  }
 
  @Test
  void testBuildJDBCUrlWithParams() {
- JsonNode expectedConfig = Jsons.deserialize(config);
- String params = "someParameter1&param2=someParameter2";
+ final JsonNode expectedConfig = Jsons.deserialize(config);
+ final String params = "someParameter1&param2=someParameter2";
  ((ObjectNode) expectedConfig).put("jdbc_url_params", params);
 
- String jdbcURL = SnowflakeDataSourceUtils.buildJDBCUrl(expectedConfig);
+ final String jdbcURL = SnowflakeDataSourceUtils.buildJDBCUrl(expectedConfig, SnowflakeDataSourceUtils.AIRBYTE_OSS);
 
  assertEquals(expectedJdbcUrl + "&" + params, jdbcURL);
  }

docs/integrations/sources/snowflake.md

@@ -120,29 +120,30 @@ To read more please check official [Snowflake documentation](https://docs.snowfl
 
 ## Changelog
 
-| Version | Date | Pull Request | Subject |
-|:--------|:-----------| :--- | :--- |
+| Version | Date | Pull Request | Subject |
+| :------ | :--------- |:---------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.1.26 | 2022-11-10 | [19314](https://github.com/airbytehq/airbyte/pull/19314) | Set application id in JDBC URL params based on OSS/Cloud environment |
 | 0.1.25 | 2022-11-10 | [15535](https://github.com/airbytehq/airbyte/pull/15535) | Update incremental query to avoid data missing when new data is inserted at the same time as a sync starts under non-CDC incremental mode |
-| 0.1.24 | 2022-09-26 | [17144](https://github.com/airbytehq/airbyte/pull/17144) | Fixed bug with incorrect date-time datatypes handling |
-| 0.1.23 | 2022-09-26 | [17116](https://github.com/airbytehq/airbyte/pull/17116) | added connection string identifier |
-| 0.1.22 | 2022-09-21 | [16766](https://github.com/airbytehq/airbyte/pull/16766) | Update JDBC Driver version to 3.13.22 |
-| 0.1.21 | 2022-09-14 | [15668](https://github.com/airbytehq/airbyte/pull/15668) | Wrap logs in AirbyteLogMessage |
-| 0.1.20 | 2022-09-01 | [16258](https://github.com/airbytehq/airbyte/pull/16258) | Emit state messages more frequently |
-| 0.1.19 | 2022-08-19 | [15797](https://github.com/airbytehq/airbyte/pull/15797) | Allow using role during oauth |
-| 0.1.18 | 2022-08-18 | [14356](https://github.com/airbytehq/airbyte/pull/14356) | DB Sources: only show a table can sync incrementally if at least one column can be used as a cursor field |
-| 0.1.17 | 2022-08-09 | [15314](https://github.com/airbytehq/airbyte/pull/15314) | Discover integer columns as integers rather than floats |
-| 0.1.16 | 2022-08-04 | [15314](https://github.com/airbytehq/airbyte/pull/15314) | (broken, do not use) Discover integer columns as integers rather than floats |
-| 0.1.15 | 2022-07-22 | [14828](https://github.com/airbytehq/airbyte/pull/14828) | Source Snowflake: Source/Destination doesn't respect DATE data type |
-| 0.1.14 | 2022-07-22 | [14714](https://github.com/airbytehq/airbyte/pull/14714) | Clarified error message when invalid cursor column selected |
-| 0.1.13 | 2022-07-14 | [14574](https://github.com/airbytehq/airbyte/pull/14574) | Removed additionalProperties:false from JDBC source connectors |
-| 0.1.12 | 2022-04-29 | [12480](https://github.com/airbytehq/airbyte/pull/12480) | Query tables with adaptive fetch size to optimize JDBC memory consumption |
-| 0.1.11 | 2022-04-27 | [10953](https://github.com/airbytehq/airbyte/pull/10953) | Implement OAuth flow |
-| 0.1.9 | 2022-02-21 | [10242](https://github.com/airbytehq/airbyte/pull/10242) | Fixed cursor for old connectors that use non-microsecond format. Now connectors work with both formats |
-| 0.1.8 | 2022-02-18 | [10242](https://github.com/airbytehq/airbyte/pull/10242) | Updated timestamp transformation with microseconds |
-| 0.1.7 | 2022-02-14 | [10256](https://github.com/airbytehq/airbyte/pull/10256) | Add `-XX:+ExitOnOutOfMemoryError` JVM option |
-| 0.1.6 | 2022-01-25 | [9623](https://github.com/airbytehq/airbyte/pull/9623) | Add jdbc_url_params support for optional JDBC parameters |
-| 0.1.5 | 2022-01-19 | [9567](https://github.com/airbytehq/airbyte/pull/9567) | Added parameter for keeping JDBC session alive |
-| 0.1.4 | 2021-12-30 | [9203](https://github.com/airbytehq/airbyte/pull/9203) | Update connector fields title/description |
-| 0.1.3 | 2021-01-11 | [9304](https://github.com/airbytehq/airbyte/pull/9304) | Upgrade version of JDBC driver |
-| 0.1.2 | 2021-10-21 | [7257](https://github.com/airbytehq/airbyte/pull/7257) | Fixed parsing of extreme values for FLOAT and NUMBER data types |
-| 0.1.1 | 2021-08-13 | [4699](https://github.com/airbytehq/airbyte/pull/4699) | Added json config validator |
+| 0.1.24 | 2022-09-26 | [17144](https://github.com/airbytehq/airbyte/pull/17144) | Fixed bug with incorrect date-time datatypes handling  |
+| 0.1.23 | 2022-09-26 | [17116](https://github.com/airbytehq/airbyte/pull/17116) | added connection string identifier  |
+| 0.1.22 | 2022-09-21 | [16766](https://github.com/airbytehq/airbyte/pull/16766) | Update JDBC Driver version to 3.13.22  |
+| 0.1.21 | 2022-09-14 | [15668](https://github.com/airbytehq/airbyte/pull/15668) | Wrap logs in AirbyteLogMessage  |
+| 0.1.20 | 2022-09-01 | [16258](https://github.com/airbytehq/airbyte/pull/16258) | Emit state messages more frequently  |
+| 0.1.19 | 2022-08-19 | [15797](https://github.com/airbytehq/airbyte/pull/15797) | Allow using role during oauth  |
+| 0.1.18 | 2022-08-18 | [14356](https://github.com/airbytehq/airbyte/pull/14356) | DB Sources: only show a table can sync incrementally if at least one column can be used as a cursor field  |
+| 0.1.17 | 2022-08-09 | [15314](https://github.com/airbytehq/airbyte/pull/15314) | Discover integer columns as integers rather than floats  |
+| 0.1.16 | 2022-08-04 | [15314](https://github.com/airbytehq/airbyte/pull/15314) | (broken, do not use) Discover integer columns as integers rather than floats  |
+| 0.1.15 | 2022-07-22 | [14828](https://github.com/airbytehq/airbyte/pull/14828) | Source Snowflake: Source/Destination doesn't respect DATE data type  |
+| 0.1.14 | 2022-07-22 | [14714](https://github.com/airbytehq/airbyte/pull/14714) | Clarified error message when invalid cursor column selected  |
+| 0.1.13 | 2022-07-14 | [14574](https://github.com/airbytehq/airbyte/pull/14574) | Removed additionalProperties:false from JDBC source connectors  |
+| 0.1.12 | 2022-04-29 | [12480](https://github.com/airbytehq/airbyte/pull/12480) | Query tables with adaptive fetch size to optimize JDBC memory consumption  |
+| 0.1.11 | 2022-04-27 | [10953](https://github.com/airbytehq/airbyte/pull/10953) | Implement OAuth flow  |
+| 0.1.9 | 2022-02-21 | [10242](https://github.com/airbytehq/airbyte/pull/10242) | Fixed cursor for old connectors that use non-microsecond format. Now connectors work with both formats  |
+| 0.1.8 | 2022-02-18 | [10242](https://github.com/airbytehq/airbyte/pull/10242) | Updated timestamp transformation with microseconds  |
+| 0.1.7 | 2022-02-14 | [10256](https://github.com/airbytehq/airbyte/pull/10256) | Add `-XX:+ExitOnOutOfMemoryError` JVM option  |
+| 0.1.6 | 2022-01-25 | [9623](https://github.com/airbytehq/airbyte/pull/9623)  | Add jdbc_url_params support for optional JDBC parameters  |
+| 0.1.5 | 2022-01-19 | [9567](https://github.com/airbytehq/airbyte/pull/9567)  | Added parameter for keeping JDBC session alive  |
+| 0.1.4 | 2021-12-30 | [9203](https://github.com/airbytehq/airbyte/pull/9203)  | Update connector fields title/description  |
+| 0.1.3 | 2021-01-11 | [9304](https://github.com/airbytehq/airbyte/pull/9304)  | Upgrade version of JDBC driver  |
+| 0.1.2 | 2021-10-21 | [7257](https://github.com/airbytehq/airbyte/pull/7257)  | Fixed parsing of extreme values for FLOAT and NUMBER data types  |
+| 0.1.1 | 2021-08-13 | [4699](https://github.com/airbytehq/airbyte/pull/4699)  | Added json config validator  |