Version Normalization as if it were part of core

.github/workflows/gradle.yml

@@ -89,6 +89,7 @@ jobs:
         run: |
           docker login -u airbytebot -p ${DOCKER_PASSWORD}
           docker-compose -f docker-compose.build.yaml push
+          docker-compose -f docker-compose.normalization.build.yaml push
         env:
           DOCKER_PASSWORD: ${{ secrets.DOCKER_PASSWORD }}
 

airbyte-integrations/bases/base-normalization/Dockerfile

@@ -15,6 +15,3 @@ COPY dbt-project-template/ ./dbt-template/
 WORKDIR /airbyte
 
 ENTRYPOINT ["/airbyte/entrypoint.sh"]
-
-LABEL io.airbyte.version=0.1.1
-LABEL io.airbyte.name=airbyte/normalization

airbyte-integrations/bases/base-normalization/build.gradle

@@ -1,5 +1,4 @@
 plugins {
-    id 'airbyte-docker'
     id 'airbyte-python'
 }
 

airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java

@@ -386,7 +386,7 @@ private void runSync(JsonNode config, List<AirbyteMessage> messages, ConfiguredA
       return;
     }
 
-    final NormalizationRunner runner = NormalizationRunnerFactory.create(
+    final NormalizationRunner runner = new NormalizationRunnerFactory("dev").create(
         getImageName(),
         pbf, targetConfig.getDestinationConnectionConfiguration());
     runner.start();

airbyte-integrations/connectors/destination-bigquery/build.gradle

@@ -18,7 +18,8 @@ dependencies {
     implementation project(':airbyte-queue')
 
     integrationTestImplementation project(':airbyte-integrations:bases:standard-destination-test')
-    integrationTestImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)
+    // depends on normalization docker image creation, which is handled in composeBuild.
+    integrationTestImplementation files(task('composeBuild').outputs)
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
 }

airbyte-integrations/connectors/destination-jdbc/build.gradle

@@ -20,5 +20,6 @@ dependencies {
     integrationTestImplementation "org.testcontainers:postgresql:1.15.0-rc2"
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
-    integrationTestImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)
+    // depends on normalization docker image creation, which is handled in composeBuild.
+    integrationTestImplementation files(task('composeBuild').outputs)
 }

airbyte-integrations/connectors/destination-postgres/build.gradle

@@ -21,5 +21,6 @@ dependencies {
     integrationTestImplementation "org.testcontainers:postgresql:1.15.0-rc2"
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
-    integrationTestImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)
+    // depends on normalization docker image creation, which is handled in composeBuild.
+    integrationTestImplementation files(task('composeBuild').outputs)
 }

airbyte-integrations/connectors/destination-redshift/build.gradle

@@ -32,5 +32,6 @@ dependencies {
     integrationTestImplementation project(':airbyte-integrations:bases:standard-destination-test')
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
-    integrationTestImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)
+    // depends on normalization docker image creation, which is handled in composeBuild.
+    integrationTestImplementation files(task('composeBuild').outputs)
 }

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

@@ -17,7 +17,8 @@ dependencies {
 
     integrationTestImplementation project(':airbyte-integrations:bases:standard-destination-test')
     integrationTestImplementation project(':airbyte-integrations:connectors:destination-snowflake')
-    integrationTestImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs)
+    // depends on normalization docker image creation, which is handled in composeBuild.
+    integrationTestImplementation files(task('composeBuild').outputs)
     integrationTestImplementation 'org.apache.commons:commons-lang3:3.11'
 
     implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)

airbyte-scheduler/src/main/java/io/airbyte/scheduler/SchedulerApp.java

@@ -36,6 +36,7 @@
 import io.airbyte.db.Databases;
 import io.airbyte.scheduler.persistence.DefaultJobPersistence;
 import io.airbyte.scheduler.persistence.JobPersistence;
+import io.airbyte.workers.normalization.NormalizationRunnerFactory;
 import io.airbyte.workers.process.DockerProcessBuilderFactory;
 import io.airbyte.workers.process.ProcessBuilderFactory;
 import java.nio.file.Path;
@@ -68,21 +69,24 @@ public class SchedulerApp {
   private final ProcessBuilderFactory pbf;
   private final JobPersistence jobPersistence;
   private final ConfigRepository configRepository;
+  private final String airbyteVersion;
 
   public SchedulerApp(Path workspaceRoot,
                       ProcessBuilderFactory pbf,
                       JobPersistence jobPersistence,
-                      ConfigRepository configRepository) {
+                      ConfigRepository configRepository,
+                      String airbyteVersion) {
     this.workspaceRoot = workspaceRoot;
     this.pbf = pbf;
     this.jobPersistence = jobPersistence;
     this.configRepository = configRepository;
+    this.airbyteVersion = airbyteVersion;
   }
 
   public void start() {
     final ExecutorService workerThreadPool = Executors.newFixedThreadPool(MAX_WORKERS, THREAD_FACTORY);
     final ScheduledExecutorService scheduledPool = Executors.newSingleThreadScheduledExecutor();
-    final WorkerRunFactory workerRunFactory = new WorkerRunFactory(workspaceRoot, pbf);
+    final WorkerRunFactory workerRunFactory = new WorkerRunFactory(workspaceRoot, pbf, new NormalizationRunnerFactory(airbyteVersion));
 
     final JobRetrier jobRetrier = new JobRetrier(jobPersistence, Instant::now);
     final JobScheduler jobScheduler = new JobScheduler(jobPersistence, configRepository);
@@ -129,7 +133,7 @@ public static void main(String[] args) {
     TrackingClientSingleton.initialize(configs.getTrackingStrategy(), configs.getAirbyteVersion(), configRepository);
 
     LOGGER.info("Launching scheduler...");
-    new SchedulerApp(workspaceRoot, pbf, jobPersistence, configRepository).start();
+    new SchedulerApp(workspaceRoot, pbf, jobPersistence, configRepository, configs.getAirbyteVersion()).start();
   }
 
 }

airbyte-scheduler/src/main/java/io/airbyte/scheduler/WorkerRunFactory.java

@@ -65,18 +65,22 @@ public class WorkerRunFactory {
   private final Path workspaceRoot;
   private final ProcessBuilderFactory pbf;
   private final Creator creator;
+  private final NormalizationRunnerFactory normalizationRunnerFactory;
 
   public WorkerRunFactory(final Path workspaceRoot,
-                          final ProcessBuilderFactory pbf) {
-    this(workspaceRoot, pbf, WorkerRun::new);
+                          final ProcessBuilderFactory pbf,
+                          final NormalizationRunnerFactory normalizationRunnerFactory) {
+    this(workspaceRoot, pbf, WorkerRun::new, normalizationRunnerFactory);
   }
 
   WorkerRunFactory(final Path workspaceRoot,
                    final ProcessBuilderFactory pbf,
-                   final Creator creator) {
+                   final Creator creator,
+                   final NormalizationRunnerFactory normalizationRunnerFactory) {
     this.workspaceRoot = workspaceRoot;
     this.pbf = pbf;
     this.creator = creator;
+    this.normalizationRunnerFactory = normalizationRunnerFactory;
   }
 
   public WorkerRun create(final Job job) {
@@ -141,7 +145,7 @@ private WorkerRun createSyncWorker(JobSyncConfig config, Path jobRoot) {
                 new DefaultAirbyteSource(sourceLauncher),
                 new DefaultAirbyteDestination(destinationLauncher),
                 new AirbyteMessageTracker(),
-                NormalizationRunnerFactory.create(
+                normalizationRunnerFactory.create(
                     config.getDestinationDockerImage(),
                     pbf,
                     syncInput.getDestinationConnection().getConfiguration()))));

airbyte-scheduler/src/test/java/io/airbyte/scheduler/WorkerRunFactoryTest.java

@@ -40,6 +40,7 @@
 import io.airbyte.config.StandardDiscoverCatalogInput;
 import io.airbyte.config.StandardSyncInput;
 import io.airbyte.workers.Worker;
+import io.airbyte.workers.normalization.NormalizationRunnerFactory;
 import io.airbyte.workers.process.ProcessBuilderFactory;
 import io.airbyte.workers.wrappers.JobOutputCheckConnectionWorker;
 import io.airbyte.workers.wrappers.JobOutputDiscoverSchemaWorker;
@@ -65,6 +66,8 @@ class WorkerRunFactoryTest {
   private WorkerRunFactory.Creator creator;
 
   private WorkerRunFactory factory;
+  private ProcessBuilderFactory pbf;
+  private NormalizationRunnerFactory normalizationRunnerFactory;
 
   @BeforeEach
   void setUp() throws IOException {
@@ -74,9 +77,11 @@ void setUp() throws IOException {
     when(job.getConfig().getSync().getDestinationDockerImage()).thenReturn("airbyte/destination-moon:0.1.0");
 
     creator = mock(WorkerRunFactory.Creator.class);
+    pbf = mock(ProcessBuilderFactory.class);
+    normalizationRunnerFactory = mock(NormalizationRunnerFactory.class);
     rootPath = Files.createTempDirectory(Files.createDirectories(TEST_ROOT), "test");
 
-    factory = new WorkerRunFactory(rootPath, mock(ProcessBuilderFactory.class), creator);
+    factory = new WorkerRunFactory(rootPath, pbf, creator, normalizationRunnerFactory);
   }
 
   @SuppressWarnings("unchecked")
@@ -114,18 +119,21 @@ void testSchema() {
   void testSync() {
     when(job.getConfig().getConfigType()).thenReturn(JobConfig.ConfigType.SYNC);
 
-    factory.create(job);
-
-    StandardSyncInput expectedInput = new StandardSyncInput()
+    final StandardSyncInput expectedInput = new StandardSyncInput()
         .withSourceConnection(job.getConfig().getSync().getSourceConnection())
         .withDestinationConnection(job.getConfig().getSync().getDestinationConnection())
         .withCatalog(AirbyteProtocolConverters.toConfiguredCatalog(job.getConfig().getSync().getStandardSync().getSchema()))
         .withConnectionId(job.getConfig().getSync().getStandardSync().getConnectionId())
         .withSyncMode(job.getConfig().getSync().getStandardSync().getSyncMode())
         .withState(job.getConfig().getSync().getState());
 
+    factory.create(job);
+
     ArgumentCaptor<Worker<StandardSyncInput, JobOutput>> argument = ArgumentCaptor.forClass(Worker.class);
     verify(creator).create(eq(rootPath.resolve("1").resolve("2")), eq(expectedInput), argument.capture());
+    verify(normalizationRunnerFactory)
+        .create(job.getConfig().getSync().getDestinationDockerImage(), pbf, expectedInput.getDestinationConnection().getConfiguration());
+
     Assertions.assertTrue(argument.getValue() instanceof JobOutputSyncWorker);
   }
 

airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java

@@ -43,10 +43,11 @@ public class DefaultNormalizationRunner implements NormalizationRunner {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DefaultNormalizationRunner.class);
 
-  public static final String NORMALIZATION_IMAGE_NAME = "airbyte/normalization:dev";
+  public static final String NORMALIZATION_IMAGE_NAME = "airbyte/normalization";
 
   private final DestinationType destinationType;
   private final ProcessBuilderFactory pbf;
+  private final String version;
 
   private Process process = null;
 
@@ -57,9 +58,10 @@ public enum DestinationType {
     SNOWFLAKE
   }
 
-  public DefaultNormalizationRunner(final DestinationType destinationType, final ProcessBuilderFactory pbf) {
+  public DefaultNormalizationRunner(final DestinationType destinationType, final ProcessBuilderFactory pbf, final String version) {
     this.destinationType = destinationType;
     this.pbf = pbf;
+    this.version = version;
   }
 
   @Override
@@ -68,7 +70,7 @@ public boolean normalize(Path jobRoot, JsonNode config, ConfiguredAirbyteCatalog
     IOs.writeFile(jobRoot, WorkerConstants.CATALOG_JSON_FILENAME, Jsons.serialize(catalog));
 
     try {
-      process = pbf.create(jobRoot, NORMALIZATION_IMAGE_NAME, "run",
+      process = pbf.create(jobRoot, NORMALIZATION_IMAGE_NAME + ":" + version, "run",
           "--integration-type", destinationType.toString().toLowerCase(),
           "--config", WorkerConstants.TARGET_CONFIG_JSON_FILENAME,
           "--catalog", WorkerConstants.CATALOG_JSON_FILENAME).start();

airbyte-workers/src/main/java/io/airbyte/workers/normalization/NormalizationRunnerFactory.java

@@ -42,15 +42,21 @@ public class NormalizationRunnerFactory {
           .put("airbyte/destination-snowflake", DefaultNormalizationRunner.DestinationType.SNOWFLAKE)
           .build();
 
-  public static NormalizationRunner create(String imageName, ProcessBuilderFactory pbf, JsonNode config) {
+  private final String normalizationVersion;
+
+  public NormalizationRunnerFactory(String normalizationVersion) {
+    this.normalizationVersion = normalizationVersion;
+  }
+
+  public NormalizationRunner create(String imageName, ProcessBuilderFactory pbf, JsonNode config) {
     if (!shouldNormalize(config)) {
       return new NoOpNormalizationRunner();
     }
 
     final String imageNameWithoutTag = imageName.split(":")[0];
 
     if (NORMALIZATION_MAPPING.containsKey(imageNameWithoutTag)) {
-      return new DefaultNormalizationRunner(NORMALIZATION_MAPPING.get(imageNameWithoutTag), pbf);
+      return new DefaultNormalizationRunner(NORMALIZATION_MAPPING.get(imageNameWithoutTag), pbf, normalizationVersion);
     } else {
       throw new IllegalStateException(
           String.format("Requested normalization for %s, but it is not included in the normalization mapping.", imageName));

airbyte-workers/src/test/java/io/airbyte/workers/normalization/DefaultNormalizationRunnerTest.java

@@ -45,6 +45,8 @@
 
 class DefaultNormalizationRunnerTest {
 
+  private static final String NORMALIZATION_IMAGE_VERSION = "organic";
+
   private Path jobRoot;
   private ProcessBuilderFactory pbf;
   private Process process;
@@ -61,7 +63,7 @@ void setup() throws IOException, WorkerException {
     config = mock(JsonNode.class);
     catalog = mock(ConfiguredAirbyteCatalog.class);
 
-    when(pbf.create(jobRoot, DefaultNormalizationRunner.NORMALIZATION_IMAGE_NAME, "run",
+    when(pbf.create(jobRoot, DefaultNormalizationRunner.NORMALIZATION_IMAGE_NAME + ":" + NORMALIZATION_IMAGE_VERSION, "run",
         "--integration-type", "bigquery",
         "--config", WorkerConstants.TARGET_CONFIG_JSON_FILENAME,
         "--catalog", WorkerConstants.CATALOG_JSON_FILENAME))
@@ -73,7 +75,7 @@ void setup() throws IOException, WorkerException {
 
   @Test
   void test() throws Exception {
-    final NormalizationRunner runner = new DefaultNormalizationRunner(DestinationType.BIGQUERY, pbf);
+    final NormalizationRunner runner = new DefaultNormalizationRunner(DestinationType.BIGQUERY, pbf, NORMALIZATION_IMAGE_VERSION);
 
     when(process.exitValue()).thenReturn(0);
 
@@ -84,7 +86,7 @@ void test() throws Exception {
   public void testClose() throws Exception {
     when(process.isAlive()).thenReturn(true).thenReturn(false);
 
-    final NormalizationRunner runner = new DefaultNormalizationRunner(DestinationType.BIGQUERY, pbf);
+    final NormalizationRunner runner = new DefaultNormalizationRunner(DestinationType.BIGQUERY, pbf, NORMALIZATION_IMAGE_VERSION);
     runner.normalize(jobRoot, config, catalog);
     runner.close();
 
@@ -95,7 +97,7 @@ public void testClose() throws Exception {
   public void testFailure() {
     doThrow(new RuntimeException()).when(process).exitValue();
 
-    final NormalizationRunner runner = new DefaultNormalizationRunner(DestinationType.BIGQUERY, pbf);
+    final NormalizationRunner runner = new DefaultNormalizationRunner(DestinationType.BIGQUERY, pbf, NORMALIZATION_IMAGE_VERSION);
     assertThrows(RuntimeException.class, () -> runner.normalize(jobRoot, config, catalog));
 
     verify(process).destroy();

airbyte-workers/src/test/java/io/airbyte/workers/normalization/NormalizationRunnerFactoryTest.java

@@ -43,30 +43,32 @@ class NormalizationRunnerFactoryTest {
 
   private static final JsonNode CONFIG_WITH_NORMALIZATION = Jsons.jsonNode(ImmutableMap.of("basic_normalization", true));
   private ProcessBuilderFactory pbf;
+  private NormalizationRunnerFactory normalizationRunnerFactory;
 
   @BeforeEach
   void setup() {
     pbf = mock(ProcessBuilderFactory.class);
+    normalizationRunnerFactory = new NormalizationRunnerFactory("organic");
   }
 
   @Test
   void testMappings() {
     assertEquals(DestinationType.BIGQUERY,
-        ((DefaultNormalizationRunner) NormalizationRunnerFactory.create(
+        ((DefaultNormalizationRunner) normalizationRunnerFactory.create(
             "airbyte/destination-bigquery:0.1.0", pbf, CONFIG_WITH_NORMALIZATION)).getDestinationType());
     assertEquals(DestinationType.POSTGRES,
-        ((DefaultNormalizationRunner) NormalizationRunnerFactory.create(
+        ((DefaultNormalizationRunner) normalizationRunnerFactory.create(
             "airbyte/destination-postgres:0.1.0", pbf, CONFIG_WITH_NORMALIZATION)).getDestinationType());
     assertEquals(DestinationType.SNOWFLAKE,
-        ((DefaultNormalizationRunner) NormalizationRunnerFactory.create(
+        ((DefaultNormalizationRunner) normalizationRunnerFactory.create(
             "airbyte/destination-snowflake:0.1.0", pbf, CONFIG_WITH_NORMALIZATION)).getDestinationType());
     assertThrows(IllegalStateException.class,
-        () -> NormalizationRunnerFactory.create("airbyte/destination-csv:0.1.0", pbf, CONFIG_WITH_NORMALIZATION));
+        () -> normalizationRunnerFactory.create("airbyte/destination-csv:0.1.0", pbf, CONFIG_WITH_NORMALIZATION));
   }
 
   @Test
   void testShouldNotNormalize() {
-    assertTrue(NormalizationRunnerFactory.create("airbyte/destination-bigquery:0.1.0", pbf,
+    assertTrue(normalizationRunnerFactory.create("airbyte/destination-bigquery:0.1.0", pbf,
         Jsons.jsonNode(Collections.emptyMap())) instanceof NoOpNormalizationRunner);
   }
 

build.gradle

@@ -91,15 +91,20 @@ allprojects {
     apply plugin: 'base'
 
     afterEvaluate { project ->
+        // format is '{project.group}.{project.name}'.
+        // note: project.group returns a period delimited the fully qualified package name.
         def composeDeps = [
-                "airbyte-config:init",
-                "airbyte-db",
-                "airbyte-scheduler",
-                "airbyte-server",
-                "airbyte-webapp",
+                'io.airbyte.airbyte-config.init',
+                'io.airbyte.airbyte-db',
+                'io.airbyte.airbyte-scheduler',
+                'io.airbyte.airbyte-server',
+                'io.airbyte.airbyte-webapp',
+                'io.airbyte.airbyte-integrations.bases.base-normalization'
         ].toSet().asImmutable()
 
-        if (project.name in composeDeps) {
+
+        def fullProjectName = (project.group + "." + project.name).toString()
+        if (fullProjectName in composeDeps) {
             composeBuild.dependsOn(project.tasks.assemble)
         }
     }
@@ -194,7 +199,7 @@ task composeBuild {
     doFirst {
         exec {
             workingDir rootDir
-            commandLine 'docker-compose', '-f', 'docker-compose.build.yaml', 'build', '--parallel', '--quiet'
+            commandLine 'docker-compose', '-f', 'docker-compose.build.yaml', '-f', 'docker-compose.normalization.build.yaml', 'build', '--parallel', '--quiet'
         }
     }
 }

docker-compose.normalization.build.yaml

@@ -0,0 +1,10 @@
+version: "3.7"
+
+services:
+  normalization:
+    image: airbyte/normalization:dev
+    build:
+      dockerfile: Dockerfile
+      context: airbyte-integrations/bases/base-normalization
+      labels:
+        io.airbyte.git-revision: ${GIT_REVISION}