Revert "Debug migration acceptance test"

This reverts commit 3fd8f2b.

Author: tuliren

airbyte-migration/src/main/java/io/airbyte/migrate/Migrate.java

@@ -36,6 +36,7 @@
 import io.airbyte.commons.version.AirbyteVersion;
 import io.airbyte.commons.yaml.Yamls;
 import io.airbyte.validation.json.JsonSchemaValidator;
+import io.airbyte.validation.json.JsonValidationException;
 import java.io.BufferedWriter;
 import java.io.File;
 import java.io.FileWriter;
@@ -108,9 +109,8 @@ public void run(MigrateConfig migrateConfig) throws IOException {
       // run migration
       // write output of each migration to disk.
       final Migration migration = migrations.get(i);
-      LOGGER.info("Migrating from {} to {}", migrations.get(i - 1).getVersion(), migration.getVersion());
+      LOGGER.info("Migrating from version: {} to version {}.", migrations.get(i - 1).getVersion(), migration.getVersion());
       final Path outputPath = runMigration(migration, inputPath);
-      LOGGER.info("Migration completed from {} to {}", migrations.get(i - 1).getVersion(), migration.getVersion());
       IOs.writeFile(outputPath.resolve(VERSION_FILE_NAME), migration.getVersion());
       inputPath = outputPath;
     }
@@ -131,7 +131,15 @@ private Path runMigration(Migration migration, Path migrationInputRoot) throws I
     final Map<ResourceId, Stream<JsonNode>> inputDataStreams = inputData.entrySet().stream()
         .collect(Collectors.toMap(
             Map.Entry::getKey,
-            entry -> MoreStreams.toStream(entry.getValue())));
+            entry -> MoreStreams.toStream(entry.getValue())
+                .peek(r -> {
+                  try {
+                    jsonSchemaValidator.ensure(migration.getInputSchema().get(entry.getKey()), r);
+                  } catch (JsonValidationException e) {
+                    throw new IllegalArgumentException(
+                        String.format("Input data schema does not match declared input schema %s.", entry.getKey().getName()), e);
+                  }
+                })));
 
     final Map<ResourceId, RecordConsumer> outputStreams = createOutputStreams(migration, tmpOutputDir);
     // make the java compiler happy (it can't resolve that RecordConsumer is, in fact, a

airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/DefaultJobPersistence.java

@@ -636,8 +636,6 @@ private Stream<JsonNode> exportTable(final String schema, final String tableName
             .map(Field::getName)
             .collect(Collectors.toSet());
         final JsonNode row = Jsons.deserialize(record.formatJSON(DB_JSON_FORMAT));
-        System.out.println("Json row: " + row);
-        LOGGER.info("Json row: {}", row);
         // for json fields, deserialize them so they are treated as objects instead of strings. this is to
         // get around that formatJson doesn't handle deserializing them for us.
         jsonFieldNames.forEach(jsonFieldName -> ((ObjectNode) row).replace(jsonFieldName, Jsons.deserialize(row.get(jsonFieldName).asText())));

airbyte-tests/src/automaticMigrationAcceptanceTest/java/io/airbyte/test/automaticMigrationAcceptance/MigrationAcceptanceTest.java

@@ -51,12 +51,14 @@
 import io.airbyte.commons.version.AirbyteVersion;
 import io.airbyte.test.airbyte_test_container.AirbyteTestContainer;
 import java.io.File;
+import java.io.FileInputStream;
 import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.function.Consumer;
 import org.junit.jupiter.api.Test;
@@ -77,15 +79,29 @@ public class MigrationAcceptanceTest {
 
   @Test
   public void testAutomaticMigration() throws Exception {
-    final String targetVersion = "0.29.16-alpha";
+    // default to version in env file but can override it.
+    final String targetVersion;
+    if (System.getenv("MIGRATION_TEST_VERSION") != null) {
+      targetVersion = System.getenv("MIGRATION_TEST_VERSION");
+    } else {
+      final Properties prop = new Properties();
+      prop.load(new FileInputStream(ENV_FILE));
+      targetVersion = prop.getProperty("VERSION");
+    }
     LOGGER.info("Using version: {} as target version", targetVersion);
 
     firstRun();
     secondRun(targetVersion);
   }
 
   private Consumer<String> logConsumerForServer(Set<String> expectedLogs) {
-    return logLine -> expectedLogs.removeIf(logLine::contains);
+    return logLine -> expectedLogs.removeIf(entry -> {
+      if (logLine.contains("Migrating from version")) {
+        System.out.println("logLine = " + logLine);
+        System.out.println("logLine = " + logLine);
+      }
+      return logLine.contains(entry);
+    });
   }
 
   @SuppressWarnings("UnstableApiUsage")
@@ -117,6 +133,7 @@ private String targetVersionWithoutPatch(String targetVersion) {
     return AirbyteVersion.versionWithoutPatch(targetVersion).getVersion();
   }
 
+  @SuppressWarnings("UnstableApiUsage")
   private void secondRun(String targetVersion) throws Exception {
     final Set<String> logsToExpect = new HashSet<>();
     logsToExpect.add("Version: " + targetVersion);
@@ -128,16 +145,14 @@ private void secondRun(String targetVersion) throws Exception {
     logsToExpect.add("Migrating from version: 0.22.0-alpha to version 0.23.0-alpha.");
     logsToExpect.add("Migrations complete. Now on version: " + targetVersionWithoutPatch(targetVersion));
 
-    final File dockerComposeFile = Path.of(System.getProperty("user.dir")).getParent().resolve("docker-compose.yaml").toFile();
-    final AirbyteTestContainer airbyteTestContainer = new AirbyteTestContainer.Builder(dockerComposeFile)
+    final AirbyteTestContainer airbyteTestContainer = new AirbyteTestContainer.Builder(new File(Resources.getResource("docker-compose.yaml").toURI()))
         .setEnv(ENV_FILE)
         // override to use test mounts.
         .setEnvVariable("DATA_DOCKER_MOUNT", "airbyte_data_migration_test")
         .setEnvVariable("DB_DOCKER_MOUNT", "airbyte_db_migration_test")
         .setEnvVariable("WORKSPACE_DOCKER_MOUNT", "airbyte_workspace_migration_test")
         .setEnvVariable("LOCAL_ROOT", "/tmp/airbyte_local_migration_test")
         .setEnvVariable("LOCAL_DOCKER_MOUNT", "/tmp/airbyte_local_migration_test")
-        .setEnvVariable("VERSION", targetVersion)
         .setLogListener("server", logConsumerForServer(logsToExpect))
         .build();