Remove NormalizationRunnerFactory

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

@@ -142,7 +142,7 @@ void testBootloaderAppBlankDb() throws Exception {
       val configsMigrator = new ConfigsDatabaseMigrator(configDatabase, configsFlyway);
       // this line should change with every new migration
       // to show that you meant to make a new migration to the prod database
-      assertEquals("0.40.23.001", configsMigrator.getLatestMigration().getVersion().getVersion());
+      assertEquals("0.40.23.002", configsMigrator.getLatestMigration().getVersion().getVersion());
 
       val jobsPersistence = new DefaultJobPersistence(jobDatabase);
       assertEquals(VERSION_0330_ALPHA, jobsPersistence.getVersion().get());

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

@@ -5,7 +5,6 @@
 package io.airbyte.workers.normalization;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.io.IOs;
@@ -46,7 +45,7 @@ public class DefaultNormalizationRunner implements NormalizationRunner {
       .setLogPrefix("normalization")
       .setPrefixColor(Color.GREEN_BACKGROUND);
 
-  private final DestinationType destinationType;
+  private final String normalizationIntegrationType;
   private final ProcessFactory processFactory;
   private final String normalizationImageName;
   private final NormalizationAirbyteStreamFactory streamFactory = new NormalizationAirbyteStreamFactory(CONTAINER_LOG_MDC_BUILDER);
@@ -55,31 +54,12 @@ public class DefaultNormalizationRunner implements NormalizationRunner {
 
   private Process process = null;
 
-  public enum DestinationType {
-    BIGQUERY,
-    MSSQL,
-    MYSQL,
-    ORACLE,
-    POSTGRES,
-    REDSHIFT,
-    SNOWFLAKE,
-    CLICKHOUSE,
-    TIDB
-  }
-
-  public DefaultNormalizationRunner(final DestinationType destinationType,
-                                    final ProcessFactory processFactory,
-                                    final String normalizationImageName) {
-    this.destinationType = destinationType;
-    this.processFactory = processFactory;
-    this.normalizationImageName = normalizationImageName;
-  }
-
   public DefaultNormalizationRunner(final ProcessFactory processFactory,
-                                    final String normalizationImage) {
+                                    final String normalizationImage,
+                                    final String normalizationIntegrationType) {
     this.processFactory = processFactory;
     this.normalizationImageName = normalizationImage;
-    this.destinationType = null;
+    this.normalizationIntegrationType = normalizationIntegrationType;
   }
 
   @Override
@@ -99,12 +79,12 @@ public boolean configureDbt(final String jobId,
     final String gitRepoBranch = dbtConfig.getGitRepoBranch();
     if (Strings.isNullOrEmpty(gitRepoBranch)) {
       return runProcess(jobId, attempt, jobRoot, files, resourceRequirements, "configure-dbt",
-          "--integration-type", destinationType.toString().toLowerCase(),
+          "--integration-type", normalizationIntegrationType.toLowerCase(),
           "--config", WorkerConstants.DESTINATION_CONFIG_JSON_FILENAME,
           "--git-repo", gitRepoUrl);
     } else {
       return runProcess(jobId, attempt, jobRoot, files, resourceRequirements, "configure-dbt",
-          "--integration-type", destinationType.toString().toLowerCase(),
+          "--integration-type", normalizationIntegrationType.toLowerCase(),
           "--config", WorkerConstants.DESTINATION_CONFIG_JSON_FILENAME,
           "--git-repo", gitRepoUrl,
           "--git-branch", gitRepoBranch);
@@ -124,7 +104,7 @@ public boolean normalize(final String jobId,
         WorkerConstants.DESTINATION_CATALOG_JSON_FILENAME, Jsons.serialize(catalog));
 
     return runProcess(jobId, attempt, jobRoot, files, resourceRequirements, "run",
-        "--integration-type", destinationType.toString().toLowerCase(),
+        "--integration-type", normalizationIntegrationType.toLowerCase(),
         "--config", WorkerConstants.DESTINATION_CONFIG_JSON_FILENAME,
         "--catalog", WorkerConstants.DESTINATION_CATALOG_JSON_FILENAME);
   }
@@ -233,9 +213,4 @@ private String buildInternalErrorMessageFromDbtStackTrace() {
     return errorMap.get(SentryExceptionHelper.ERROR_MAP_KEYS.ERROR_MAP_MESSAGE_KEY);
   }
 
-  @VisibleForTesting
-  DestinationType getDestinationType() {
-    return destinationType;
-  }
-
 }

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

@@ -15,6 +15,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.docker.DockerUtils;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.logging.LoggingHelper.Color;
@@ -26,7 +27,6 @@
 import io.airbyte.workers.WorkerConfigs;
 import io.airbyte.workers.WorkerConstants;
 import io.airbyte.workers.exception.WorkerException;
-import io.airbyte.workers.normalization.DefaultNormalizationRunner.DestinationType;
 import io.airbyte.workers.process.AirbyteIntegrationLauncher;
 import io.airbyte.workers.process.ProcessFactory;
 import java.io.ByteArrayInputStream;
@@ -49,6 +49,10 @@ class DefaultNormalizationRunnerTest {
   private static final String JOB_ID = "0";
   private static final int JOB_ATTEMPT = 0;
 
+  private static final String NORMALIZATION_IMAGE = "airbyte/normalization";
+  private static final String NORMALIZATION_TAG = "42.42.42";
+  private static final String INTEGRATION_TYPE = "postgres";
+
   private static Path logJobRoot;
 
   static {
@@ -82,14 +86,14 @@ void setup() throws IOException, WorkerException {
         WorkerConstants.DESTINATION_CATALOG_JSON_FILENAME, Jsons.serialize(catalog));
 
     when(processFactory.create(AirbyteIntegrationLauncher.NORMALIZE_STEP, JOB_ID, JOB_ATTEMPT, jobRoot,
-        NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME, false, false, files, null,
+        DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), false, false, files, null,
         workerConfigs.getResourceRequirements(),
         Map.of(AirbyteIntegrationLauncher.JOB_TYPE, AirbyteIntegrationLauncher.SYNC_JOB, AirbyteIntegrationLauncher.SYNC_STEP,
             AirbyteIntegrationLauncher.NORMALIZE_STEP),
         Map.of(),
         Map.of(),
         "run",
-        "--integration-type", "bigquery",
+        "--integration-type", INTEGRATION_TYPE,
         "--config", WorkerConstants.DESTINATION_CONFIG_JSON_FILENAME,
         "--catalog", WorkerConstants.DESTINATION_CATALOG_JSON_FILENAME))
             .thenReturn(process);
@@ -110,8 +114,7 @@ public void tearDown() throws IOException {
   @Test
   void test() throws Exception {
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
 
     when(process.exitValue()).thenReturn(0);
 
@@ -122,8 +125,7 @@ void test() throws Exception {
   void testLog() throws Exception {
 
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
 
     when(process.exitValue()).thenReturn(0);
 
@@ -145,8 +147,7 @@ void testClose() throws Exception {
     when(process.isAlive()).thenReturn(true).thenReturn(false);
 
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
     runner.normalize(JOB_ID, JOB_ATTEMPT, jobRoot, config, catalog, workerConfigs.getResourceRequirements());
     runner.close();
 
@@ -158,8 +159,7 @@ void testFailure() throws Exception {
     when(process.exitValue()).thenReturn(1);
 
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
     assertFalse(runner.normalize(JOB_ID, JOB_ATTEMPT, jobRoot, config, catalog, workerConfigs.getResourceRequirements()));
 
     verify(process).waitFor();
@@ -180,8 +180,7 @@ void testFailureWithTraceMessage() throws Exception {
     when(process.getInputStream()).thenReturn(new ByteArrayInputStream(errorTraceString.getBytes(StandardCharsets.UTF_8)));
 
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
     assertFalse(runner.normalize(JOB_ID, JOB_ATTEMPT, jobRoot, config, catalog, workerConfigs.getResourceRequirements()));
 
     assertEquals(1, runner.getTraceMessages().count());
@@ -207,8 +206,7 @@ void testFailureWithDbtError() throws Exception {
     when(process.getInputStream()).thenReturn(new ByteArrayInputStream(dbtErrorString.getBytes(StandardCharsets.UTF_8)));
 
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
     assertFalse(runner.normalize(JOB_ID, JOB_ATTEMPT, jobRoot, config, catalog, workerConfigs.getResourceRequirements()));
 
     assertEquals(1, runner.getTraceMessages().count());
@@ -229,8 +227,7 @@ void testFailureWithDbtErrorJsonFormat() throws Exception {
     when(process.getInputStream()).thenReturn(new ByteArrayInputStream(dbtErrorString.getBytes(StandardCharsets.UTF_8)));
 
     final NormalizationRunner runner =
-        new DefaultNormalizationRunner(DestinationType.BIGQUERY, processFactory,
-            NormalizationRunnerFactory.BASE_NORMALIZATION_IMAGE_NAME);
+        new DefaultNormalizationRunner(processFactory, DockerUtils.getTaggedImageName(NORMALIZATION_IMAGE, NORMALIZATION_TAG), INTEGRATION_TYPE);
     assertFalse(runner.normalize(JOB_ID, JOB_ATTEMPT, jobRoot, config, catalog, workerConfigs.getResourceRequirements()));
 
     assertEquals(1, runner.getTraceMessages().count());

airbyte-config/config-models/src/main/resources/types/NormalizationDestinationDefinitionConfig.yaml

@@ -0,0 +1,21 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/NormalizationDestinationDefinitionConfig.yaml
+title: NormalizationDestinationDefinitionConfig
+description: describes a normalization config for destination definition
+type: object
+required:
+  - normalizationRepository
+  - normalizationTag
+  - normalizationIntegrationType
+additionalProperties: true
+properties:
+  normalizationRepository:
+    type: string
+    description: an optional field indicating the name of the repository to be used for normalization. If the value of the flag is NULL - normalization is not used.
+  normalizationTag:
+    type: string
+    description: an optional field indicating the tag of the docker repository to be used for normalization.
+  normalizationIntegrationType:
+    type: string
+    description: an optional field indicating the type of integration dialect to use for normalization.

airbyte-config/config-models/src/main/resources/types/StandardDestinationDefinition.yaml

@@ -58,10 +58,8 @@ properties:
   protocolVersion:
     type: string
     description: the Airbyte Protocol version supported by the connector
-  normalizationRepository:
-    type: string
-  normalizationTag:
-    type: string
+  normalizationConfig:
+    "$ref": NormalizationDestinationDefinitionConfig.yaml
   supportsDbt:
     type: boolean
-    default: false
+    description: an optional flag indicating whether DBT is used in the normalization. If the flag value is NULL - DBT is not used.

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ActorDefinitionMigrator.java

@@ -16,10 +16,7 @@
 import io.airbyte.commons.util.MoreIterators;
 import io.airbyte.commons.version.AirbyteProtocolVersion;
 import io.airbyte.commons.version.AirbyteVersion;
-import io.airbyte.config.AirbyteConfig;
-import io.airbyte.config.ConfigSchema;
-import io.airbyte.config.StandardDestinationDefinition;
-import io.airbyte.config.StandardSourceDefinition;
+import io.airbyte.config.*;
 import io.airbyte.config.StandardSourceDefinition.SourceType;
 import io.airbyte.db.ExceptionWrappingDatabase;
 import io.airbyte.db.instance.configs.jooq.generated.enums.ActorType;
@@ -271,7 +268,9 @@ private void writeOrUpdateStandardDefinition(final DSLContext ctx,
       sourceDef.withProtocolVersion(getProtocolVersion(sourceDef.getSpec()));
       ConfigWriter.writeStandardSourceDefinition(Collections.singletonList(sourceDef), ctx);
     } else if (configType == ConfigSchema.STANDARD_DESTINATION_DEFINITION) {
-      final StandardDestinationDefinition destDef = Jsons.object(definition, StandardDestinationDefinition.class);
+      final NormalizationDestinationDefinitionConfig normalizationConfig = Jsons.object(definition, NormalizationDestinationDefinitionConfig.class);
+      final StandardDestinationDefinition destDef = Jsons.object(definition, StandardDestinationDefinition.class)
+          .withNormalizationConfig(normalizationConfig);
       destDef.withProtocolVersion(getProtocolVersion(destDef.getSpec()));
       ConfigWriter.writeStandardDestinationDefinition(Collections.singletonList(destDef), ctx);
     } else {