Add Airbyte Protocol V1 support.

.github/workflows/gradle.yml

@@ -12,7 +12,7 @@ on:
         description: "Enable or disable tmate session for debug during helm ac tests"
         type: choice
         default: 'false'
-        options: 
+        options:
           - 'true'
           - 'false'
         required: false
@@ -305,7 +305,7 @@ jobs:
         if: always()
         run: |
           gcs_bucket_name="dev-ab-ci-run-results"
-          filename=$(echo "${{ fromJSON( steps.connectors-test-results.outputs.json ).check_url }}" | sed 's@.*/@@')  
+          filename=$(echo "${{ fromJSON( steps.connectors-test-results.outputs.json ).check_url }}" | sed 's@.*/@@')
           echo "$filename"
           gsutil -h "Cache-Control:public" cp connectors_base_results.jsonl "gs://$gcs_bucket_name/oss/$filename.jsonl"
 
@@ -699,7 +699,7 @@ jobs:
         if: always()
         run: |
           gcs_bucket_name="dev-ab-ci-run-results"
-          filename=$(echo "${{ fromJSON( steps.platform-results.outputs.json ).check_url }}" | sed 's@.*/@@')  
+          filename=$(echo "${{ fromJSON( steps.platform-results.outputs.json ).check_url }}" | sed 's@.*/@@')
           echo "$filename"
           gsutil -h "Cache-Control:public" cp platform_results.jsonl "gs://$gcs_bucket_name/oss/$filename.jsonl"
 
@@ -756,7 +756,7 @@ jobs:
           github-token: ${{ env.PAT }}
           label: ${{ needs.start-platform-build-runner.outputs.label }}
           ec2-instance-id: ${{ needs.start-platform-build-runner.outputs.ec2-instance-id }}
-  
+
   ## Kube Acceptance Tests
   # Docker acceptance tests run as part of the build job.
   # In case of self-hosted EC2 errors, remove this block.
@@ -919,7 +919,7 @@ jobs:
         if: always()
         run: |
           gcs_bucket_name="dev-ab-ci-run-results"
-          filename=$(echo "${{ fromJSON( steps.kube-results.outputs.json ).check_url }}" | sed 's@.*/@@')  
+          filename=$(echo "${{ fromJSON( steps.kube-results.outputs.json ).check_url }}" | sed 's@.*/@@')
           echo "$filename"
           gsutil -h "Cache-Control:public" cp kube_results.jsonl "gs://$gcs_bucket_name/oss/$filename.jsonl"
 
@@ -1036,7 +1036,7 @@ jobs:
       - name: Fix EC-2 Runner
         run: |
           mkdir -p /actions-runner/_work/airbyte/airbyte && mkdir -p /actions-runner/_work/airbyte/airbyte/.kube
-        
+
       - name: Checkout Airbyte
         uses: actions/checkout@v2
         with:
@@ -1059,7 +1059,7 @@ jobs:
 
       - uses: actions/setup-python@v4
         with:
-          python-version: '3.9' 
+          python-version: '3.9'
 
       - uses: actions/setup-java@v1
         with:
@@ -1087,11 +1087,11 @@ jobs:
             sudo apt-get -y install tmate
           attempt_limit: 3
           attempt_delay: 2000 # in ms
-      
+
       - name: Start tmate session in background
         if: inputs.debug_mode == 'true'
         shell: bash
-        run: |  
+        run: |
           tmate -S /tmp/tmate.sock new-session -d               # Launch tmate in a headless mode
           tmate -S /tmp/tmate.sock wait tmate-ready             # Blocks until the SSH connection is established
           tmate -S /tmp/tmate.sock display -p '#{tmate_ssh}'    # Prints the SSH connection string
@@ -1163,7 +1163,7 @@ jobs:
         with:
           name: Kubernetes Logs
           path: /tmp/kubernetes_logs/*
-      
+
       - name: Upload test results to BuildPulse for flaky test detection
         if: "!cancelled()" # Run this step even when the tests fail. Skip if the workflow is cancelled.
         uses: Workshop64/buildpulse-action@main
@@ -1173,13 +1173,13 @@ jobs:
           path: "/actions-runner/_work/airbyte/airbyte/*"
           key: ${{ secrets.BUILDPULSE_ACCESS_KEY_ID }}
           secret: ${{ secrets.BUILDPULSE_SECRET_ACCESS_KEY }}
-      
+
       - name: "Display logs of k3s"
         if: failure()
         shell: bash
         run: |
           journalctl -xeu k3s.service
-    
+
 
   # # In case of self-hosted EC2 errors, remove this block.
   stop-helm-acceptance-test-runner:

airbyte-commons-protocol/build.gradle

@@ -6,6 +6,7 @@ dependencies {
     testImplementation libs.bundles.micronaut.test
 
     implementation project(':airbyte-protocol:protocol-models')
+    implementation project(':airbyte-json-validation')
 }
 
 Task publishArtifactsTask = getPublishArtifactsTask("$rootProject.ext.version", project)

airbyte-commons-protocol/src/main/java/io/airbyte/commons/protocol/AirbyteMessageMigrator.java

@@ -6,15 +6,14 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import io.airbyte.commons.protocol.migrations.AirbyteMessageMigration;
+import io.airbyte.commons.protocol.migrations.MigrationContainer;
 import io.airbyte.commons.version.Version;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import jakarta.annotation.PostConstruct;
 import jakarta.inject.Singleton;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
 
 /**
  * AirbyteProtocol Message Migrator
@@ -25,104 +24,59 @@
 @Singleton
 public class AirbyteMessageMigrator {
 
-  private final List<AirbyteMessageMigration<?, ?>> migrationsToRegister;
-  private final SortedMap<String, AirbyteMessageMigration<?, ?>> migrations = new TreeMap<>();
-  private String mostRecentMajorVersion = "";
+  private final MigrationContainer<AirbyteMessageMigration<?, ?>> migrationContainer;
 
-  public AirbyteMessageMigrator(List<AirbyteMessageMigration<?, ?>> migrations) {
-    migrationsToRegister = migrations;
-  }
-
-  public AirbyteMessageMigrator() {
-    this(Collections.emptyList());
+  public AirbyteMessageMigrator(final List<AirbyteMessageMigration<?, ?>> migrations) {
+    migrationContainer = new MigrationContainer<>(migrations);
   }
 
   @PostConstruct
   public void initialize() {
-    migrationsToRegister.forEach(this::registerMigration);
+    migrationContainer.initialize();
   }
 
   /**
    * Downgrade a message from the most recent version to the target version by chaining all the
    * required migrations
    */
-  public <PreviousVersion, CurrentVersion> PreviousVersion downgrade(final CurrentVersion message, final Version target) {
-    if (target.getMajorVersion().equals(mostRecentMajorVersion)) {
-      return (PreviousVersion) message;
-    }
-
-    Object result = message;
-    Object[] selectedMigrations = selectMigrations(target).toArray();
-    for (int i = selectedMigrations.length; i > 0; --i) {
-      result = applyDowngrade((AirbyteMessageMigration<?, ?>) selectedMigrations[i - 1], result);
-    }
-    return (PreviousVersion) result;
+  public <PreviousVersion, CurrentVersion> PreviousVersion downgrade(final CurrentVersion message,
+                                                                     final Version target,
+                                                                     final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
+    return migrationContainer.downgrade(message, target, (migration, msg) -> applyDowngrade(migration, msg, configuredAirbyteCatalog));
   }
 
   /**
    * Upgrade a message from the source version to the most recent version by chaining all the required
    * migrations
    */
-  public <PreviousVersion, CurrentVersion> CurrentVersion upgrade(final PreviousVersion message, final Version source) {
-    if (source.getMajorVersion().equals(mostRecentMajorVersion)) {
-      return (CurrentVersion) message;
-    }
-
-    Object result = message;
-    for (var migration : selectMigrations(source)) {
-      result = applyUpgrade(migration, result);
-    }
-    return (CurrentVersion) result;
+  public <PreviousVersion, CurrentVersion> CurrentVersion upgrade(final PreviousVersion message,
+                                                                  final Version source,
+                                                                  final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
+    return migrationContainer.upgrade(message, source, (migration, msg) -> applyUpgrade(migration, msg, configuredAirbyteCatalog));
   }
 
   public Version getMostRecentVersion() {
-    return new Version(mostRecentMajorVersion, "0", "0");
-  }
-
-  private Collection<AirbyteMessageMigration<?, ?>> selectMigrations(final Version version) {
-    final Collection<AirbyteMessageMigration<?, ?>> results = migrations.tailMap(version.getMajorVersion()).values();
-    if (results.isEmpty()) {
-      throw new RuntimeException("Unsupported migration version " + version.serialize());
-    }
-    return results;
+    return migrationContainer.getMostRecentVersion();
   }
 
   // Helper function to work around type casting
-  private <PreviousVersion, CurrentVersion> PreviousVersion applyDowngrade(final AirbyteMessageMigration<PreviousVersion, CurrentVersion> migration,
-                                                                           final Object message) {
-    return migration.downgrade((CurrentVersion) message);
+  private static <PreviousVersion, CurrentVersion> PreviousVersion applyDowngrade(final AirbyteMessageMigration<PreviousVersion, CurrentVersion> migration,
+                                                                                  final Object message,
+                                                                                  final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
+    return migration.downgrade((CurrentVersion) message, configuredAirbyteCatalog);
   }
 
   // Helper function to work around type casting
-  private <PreviousVersion, CurrentVersion> CurrentVersion applyUpgrade(final AirbyteMessageMigration<PreviousVersion, CurrentVersion> migration,
-                                                                        final Object message) {
-    return migration.upgrade((PreviousVersion) message);
-  }
-
-  /**
-   * Store migration in a sorted map key by the major of the lower version of the migration.
-   *
-   * The goal is to be able to retrieve the list of migrations to apply to get to/from a given
-   * version. We are only keying on the lower version because the right side (most recent version of
-   * the migration range) is always current version.
-   */
-  @VisibleForTesting
-  void registerMigration(final AirbyteMessageMigration<?, ?> migration) {
-    final String key = migration.getPreviousVersion().getMajorVersion();
-    if (!migrations.containsKey(key)) {
-      migrations.put(key, migration);
-      if (migration.getCurrentVersion().getMajorVersion().compareTo(mostRecentMajorVersion) > 0) {
-        mostRecentMajorVersion = migration.getCurrentVersion().getMajorVersion();
-      }
-    } else {
-      throw new RuntimeException("Trying to register a duplicated migration " + migration.getClass().getName());
-    }
+  private static <PreviousVersion, CurrentVersion> CurrentVersion applyUpgrade(final AirbyteMessageMigration<PreviousVersion, CurrentVersion> migration,
+                                                                               final Object message,
+                                                                               final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
+    return migration.upgrade((PreviousVersion) message, configuredAirbyteCatalog);
   }
 
   // Used for inspection of the injection
   @VisibleForTesting
   Set<String> getMigrationKeys() {
-    return migrations.keySet();
+    return migrationContainer.getMigrationKeys();
   }
 
 }

airbyte-commons-protocol/src/main/java/io/airbyte/commons/protocol/AirbyteMessageVersionedMigrator.java

@@ -5,7 +5,9 @@
 package io.airbyte.commons.protocol;
 
 import io.airbyte.commons.version.Version;
-import io.airbyte.protocol.models.v0.AirbyteMessage;
+import io.airbyte.protocol.models.AirbyteMessage;
+import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import java.util.Optional;
 
 /**
  * Wraps message migration from a fixed version to the most recent version
@@ -20,12 +22,12 @@ public AirbyteMessageVersionedMigrator(final AirbyteMessageMigrator migrator, fi
     this.version = version;
   }
 
-  public OriginalMessageType downgrade(final AirbyteMessage message) {
-    return migrator.downgrade(message, version);
+  public OriginalMessageType downgrade(final AirbyteMessage message, final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
+    return migrator.downgrade(message, version, configuredAirbyteCatalog);
   }
 
-  public AirbyteMessage upgrade(final OriginalMessageType message) {
-    return migrator.upgrade(message, version);
+  public AirbyteMessage upgrade(final OriginalMessageType message, final Optional<ConfiguredAirbyteCatalog> configuredAirbyteCatalog) {
+    return migrator.upgrade(message, version, configuredAirbyteCatalog);
   }
 
   public Version getVersion() {

airbyte-commons-protocol/src/main/java/io/airbyte/commons/protocol/AirbyteProtocolVersionedMigratorFactory.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.commons.protocol;
+
+import io.airbyte.commons.version.Version;
+import jakarta.inject.Singleton;
+
+/**
+ * Factory to build AirbyteMessageVersionedMigrator
+ */
+@Singleton
+public class AirbyteProtocolVersionedMigratorFactory {
+
+  private final AirbyteMessageMigrator airbyteMessageMigrator;
+  private final ConfiguredAirbyteCatalogMigrator configuredAirbyteCatalogMigrator;
+
+  public AirbyteProtocolVersionedMigratorFactory(final AirbyteMessageMigrator airbyteMessageMigrator,
+                                                 final ConfiguredAirbyteCatalogMigrator configuredAirbyteCatalogMigrator) {
+    this.airbyteMessageMigrator = airbyteMessageMigrator;
+    this.configuredAirbyteCatalogMigrator = configuredAirbyteCatalogMigrator;
+  }
+
+  public <T> AirbyteMessageVersionedMigrator<T> getAirbyteMessageMigrator(final Version version) {
+    return new AirbyteMessageVersionedMigrator<>(airbyteMessageMigrator, version);
+  }
+
+  public final VersionedProtocolSerializer getProtocolSerializer(final Version version) {
+    return new VersionedProtocolSerializer(configuredAirbyteCatalogMigrator, version);
+  }
+
+  public Version getMostRecentVersion() {
+    return airbyteMessageMigrator.getMostRecentVersion();
+  }
+
+}