Merge branch 'master' into grubberr/10384-source-github

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.35.30-alpha
+current_version = 0.35.31-alpha
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(\-[a-z]+)?

.env

@@ -10,7 +10,7 @@
 
 
 ### SHARED ###
-VERSION=0.35.30-alpha
+VERSION=0.35.31-alpha
 
 # When using the airbyte-db via default docker image
 CONFIG_ROOT=/data

airbyte-bootloader/Dockerfile

@@ -5,6 +5,6 @@ ENV APPLICATION airbyte-bootloader
 
 WORKDIR /app
 
-ADD bin/${APPLICATION}-0.35.30-alpha.tar /app
+ADD bin/${APPLICATION}-0.35.31-alpha.tar /app
 
-ENTRYPOINT ["/bin/bash", "-c", "${APPLICATION}-0.35.30-alpha/bin/${APPLICATION}"]
+ENTRYPOINT ["/bin/bash", "-c", "${APPLICATION}-0.35.31-alpha/bin/${APPLICATION}"]

airbyte-bootloader/src/main/java/io/airbyte/bootloader/BootloaderApp.java

@@ -55,10 +55,10 @@ public class BootloaderApp {
 
   private final Configs configs;
   private Runnable postLoadExecution;
-  private FeatureFlags featureFlags;
+  private final FeatureFlags featureFlags;
 
   @VisibleForTesting
-  public BootloaderApp(Configs configs, FeatureFlags featureFlags) {
+  public BootloaderApp(final Configs configs, final FeatureFlags featureFlags) {
     this.configs = configs;
     this.featureFlags = featureFlags;
   }
@@ -71,7 +71,7 @@ public BootloaderApp(Configs configs, FeatureFlags featureFlags) {
    * @param configs
    * @param postLoadExecution
    */
-  public BootloaderApp(Configs configs, Runnable postLoadExecution, FeatureFlags featureFlags) {
+  public BootloaderApp(final Configs configs, final Runnable postLoadExecution, final FeatureFlags featureFlags) {
     this.configs = configs;
     this.postLoadExecution = postLoadExecution;
     this.featureFlags = featureFlags;
@@ -87,7 +87,7 @@ public BootloaderApp() {
         final ConfigPersistence configPersistence = DatabaseConfigPersistence.createWithValidation(configDatabase);
         configPersistence.loadData(YamlSeedConfigPersistence.getDefault());
         LOGGER.info("Loaded seed data..");
-      } catch (IOException e) {
+      } catch (final IOException e) {
         e.printStackTrace();
       }
     };
@@ -134,7 +134,7 @@ public void load() throws Exception {
     LOGGER.info("Finished bootstrapping Airbyte environment..");
   }
 
-  public static void main(String[] args) throws Exception {
+  public static void main(final String[] args) throws Exception {
     final var bootloader = new BootloaderApp();
     bootloader.load();
   }
@@ -168,7 +168,8 @@ private static void createWorkspaceIfNoneExists(final ConfigRepository configRep
     configRepository.writeStandardWorkspace(workspace);
   }
 
-  private static void assertNonBreakingMigration(JobPersistence jobPersistence, AirbyteVersion airbyteVersion) throws IOException {
+  private static void assertNonBreakingMigration(final JobPersistence jobPersistence, final AirbyteVersion airbyteVersion)
+      throws IOException {
     // version in the database when the server main method is called. may be empty if this is the first
     // time the server is started.
     LOGGER.info("Checking illegal upgrade..");
@@ -221,7 +222,7 @@ private static void runFlywayMigration(final Configs configs, final Database con
 
   private static void cleanupZombies(final JobPersistence jobPersistence) throws IOException {
     final Configs configs = new EnvConfigs();
-    WorkflowClient wfClient =
+    final WorkflowClient wfClient =
         WorkflowClient.newInstance(WorkflowServiceStubs.newInstance(
             WorkflowServiceStubsOptions.newBuilder().setTarget(configs.getTemporalHost()).build()));
     for (final Job zombieJob : jobPersistence.listJobsWithStatus(JobStatus.RUNNING)) {

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderAppTest.java

@@ -80,7 +80,7 @@ void testBootloaderAppBlankDb() throws Exception {
         mockedConfigs.getConfigDatabaseUrl())
             .getAndInitialize();
     val configsMigrator = new ConfigsDatabaseMigrator(configDatabase, this.getClass().getName());
-    assertEquals("0.35.26.001", configsMigrator.getLatestMigration().getVersion().getVersion());
+    assertEquals("0.35.28.001", configsMigrator.getLatestMigration().getVersion().getVersion());
 
     val jobsPersistence = new DefaultJobPersistence(jobDatabase);
     assertEquals(version, jobsPersistence.getVersion().get());
@@ -112,7 +112,7 @@ void testIsLegalUpgradePredicate() {
 
   @Test
   void testPostLoadExecutionExecutes() throws Exception {
-    var testTriggered = new AtomicBoolean();
+    final var testTriggered = new AtomicBoolean();
 
     val container = new PostgreSQLContainer<>("postgres:13-alpine")
         .withDatabaseName("public")

airbyte-cdk/python/airbyte_cdk/sources/streams/core.py

@@ -171,7 +171,7 @@ def state_checkpoint_interval(self) -> Optional[int]:
         """
         return None
 
-    @deprecated(version='0.1.49', reason="You should use explicit state property instead, see IncrementalMixin docs.")
+    @deprecated(version="0.1.49", reason="You should use explicit state property instead, see IncrementalMixin docs.")
     def get_updated_state(self, current_stream_state: MutableMapping[str, Any], latest_record: Mapping[str, Any]):
         """Override to extract state from the latest record. Needed to implement incremental sync.
 

airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java

@@ -4,6 +4,9 @@
 
 package io.airbyte.commons.json;
 
+import static java.util.Collections.singletonMap;
+import static java.util.stream.Collectors.toMap;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
@@ -19,9 +22,12 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.BiConsumer;
@@ -197,6 +203,58 @@ public static int getIntOrZero(final JsonNode json, final List<String> keys) {
     return optional.map(JsonNode::asInt).orElse(0);
   }
 
+  /**
+   * Flattens an ObjectNode, or dumps it into a {null: value} map if it's not an object.
+   */
+  public static Map<String, Object> flatten(final JsonNode node) {
+    if (node.isObject()) {
+      final Map<String, Object> output = new HashMap<>();
+      for (final Iterator<Entry<String, JsonNode>> it = node.fields(); it.hasNext();) {
+        final Entry<String, JsonNode> entry = it.next();
+        final String field = entry.getKey();
+        final JsonNode value = entry.getValue();
+        mergeMaps(output, field, flatten(value));
+      }
+      return output;
+    } else {
+      final Object value;
+      if (node.isBoolean()) {
+        value = node.asBoolean();
+      } else if (node.isLong()) {
+        value = node.asLong();
+      } else if (node.isInt()) {
+        value = node.asInt();
+      } else if (node.isDouble()) {
+        value = node.asDouble();
+      } else if (node.isValueNode() && !node.isNull()) {
+        value = node.asText();
+      } else {
+        // Fallback handling for e.g. arrays
+        value = node.toString();
+      }
+      return singletonMap(null, value);
+    }
+  }
+
+  /**
+   * Prepend all keys in subMap with prefix, then merge that map into originalMap.
+   * <p>
+   * If subMap contains a null key, then instead it is replaced with prefix. I.e. {null: value} is
+   * treated as {prefix: value} when merging into originalMap.
+   */
+  public static void mergeMaps(final Map<String, Object> originalMap, final String prefix, final Map<String, Object> subMap) {
+    originalMap.putAll(subMap.entrySet().stream().collect(toMap(
+        e -> {
+          final String key = e.getKey();
+          if (key != null) {
+            return prefix + "." + key;
+          } else {
+            return prefix;
+          }
+        },
+        Entry::getValue)));
+  }
+
   /**
    * By the Jackson DefaultPrettyPrinter prints objects with an extra space as follows: {"name" :
    * "airbyte"}. We prefer {"name": "airbyte"}.

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

@@ -126,7 +126,7 @@
 - name: MySQL
   destinationDefinitionId: ca81ee7c-3163-4246-af40-094cc31e5e42
   dockerRepository: airbyte/destination-mysql
-  dockerImageTag: 0.1.16
+  dockerImageTag: 0.1.17
   documentationUrl: https://docs.airbyte.io/integrations/destinations/mysql
   icon: mysql.svg
 - name: Oracle
@@ -185,7 +185,7 @@
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.4.12
+  dockerImageTag: 0.4.14
   documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake
   icon: snowflake.svg
 - name: MariaDB ColumnStore

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

@@ -2464,7 +2464,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-mysql:0.1.16"
+- dockerImage: "airbyte/destination-mysql:0.1.17"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/mysql"
     connectionSpecification:
@@ -2515,6 +2515,13 @@
           type: "boolean"
           default: true
           order: 5
+        jdbc_url_params:
+          description: "Additional properties to pass to the JDBC URL string when\
+            \ connecting to the database formatted as 'key=value' pairs separated\
+            \ by the symbol '&'. (example: key1=value1&key2=value2&key3=value3)."
+          title: "JDBC URL Params"
+          type: "string"
+          order: 6
         tunnel_method:
           type: "object"
           title: "SSH Tunnel Method"
@@ -3818,7 +3825,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-snowflake:0.4.12"
+- dockerImage: "airbyte/destination-snowflake:0.4.14"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/snowflake"
     connectionSpecification:

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

@@ -596,10 +596,10 @@
   documentationUrl: https://docs.airbyte.io/integrations/sources/qualaroo
   icon: qualaroo.svg
   sourceType: api
-- name: Quickbooks
+- name: QuickBooks
   sourceDefinitionId: 29b409d9-30a5-4cc8-ad50-886eb846fea3
   dockerRepository: airbyte/source-quickbooks-singer
-  dockerImageTag: 0.1.4
+  dockerImageTag: 0.1.5
   documentationUrl: https://docs.airbyte.io/integrations/sources/quickbooks
   icon: qb.svg
   sourceType: api

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

@@ -6266,12 +6266,12 @@
         oauthFlowOutputParameters:
         - - "token"
         - - "key"
-- dockerImage: "airbyte/source-quickbooks-singer:0.1.4"
+- dockerImage: "airbyte/source-quickbooks-singer:0.1.5"
   spec:
-    documentationUrl: "https://docsurl.com"
+    documentationUrl: "https://docs.airbyte.com/integrations/sources/quickbooks"
     connectionSpecification:
       $schema: "http://json-schema.org/draft-07/schema#"
-      title: "Source Quickbooks Singer Spec"
+      title: "Source QuickBooks Singer Spec"
       type: "object"
       required:
       - "client_id"

airbyte-config/models/src/main/java/io/airbyte/config/ConfigSchema.java

@@ -60,6 +60,9 @@ public enum ConfigSchema implements AirbyteConfig {
 
   STANDARD_SYNC_SUMMARY("StandardSyncSummary.yaml", StandardSyncSummary.class),
 
+  ACTOR_CATALOG("ActorCatalog.yaml", ActorCatalog.class),
+  ACTOR_CATALOG_FETCH_EVENT("ActorCatalogFetchEvent.yaml", ActorCatalogFetchEvent.class),
+
   // worker
   STANDARD_SYNC_INPUT("StandardSyncInput.yaml", StandardSyncInput.class),
   NORMALIZATION_INPUT("NormalizationInput.yaml", NormalizationInput.class),