Auto Detect Schema Change environment variable (#19312)

* auto detect schema environment variable

Author: alovew

.env

@@ -113,3 +113,5 @@ METRIC_CLIENT=
 OTEL_COLLECTOR_ENDPOINT="http://host.docker.internal:4317"
 
 USE_STREAM_CAPABLE_STATE=true
+AUTO_DETECT_SCHEMA=false
+

.env.dev

@@ -28,6 +28,7 @@ SYNC_JOB_MAX_ATTEMPTS=3
 SYNC_JOB_MAX_TIMEOUT_DAYS=3
 WORKERS_MICRONAUT_ENVIRONMENTS=control-plane
 CRON_MICRONAUT_ENVIRONMENTS=control-plane
+AUTO_DETECT_SCHEMA=false
 
 # Sentry
 SENTRY_DSN=""

airbyte-commons-temporal/src/main/java/io/airbyte/commons/temporal/sync/OrchestratorConstants.java

@@ -65,7 +65,8 @@ public class OrchestratorConstants {
           EnvConfigs.STATE_STORAGE_S3_ACCESS_KEY,
           EnvConfigs.STATE_STORAGE_S3_SECRET_ACCESS_KEY,
           EnvConfigs.STATE_STORAGE_S3_REGION,
-          EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE))
+          EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE,
+          EnvVariableFeatureFlags.AUTO_DETECT_SCHEMA))
       .build();
 
   public static final String INIT_FILE_ENV_MAP = "envMap.json";

airbyte-commons-worker/src/main/java/io/airbyte/workers/process/AirbyteIntegrationLauncher.java

@@ -212,7 +212,8 @@ private Map<String, String> getWorkerMetadata() {
         WorkerEnvConstants.WORKER_CONNECTOR_IMAGE, imageName,
         WorkerEnvConstants.WORKER_JOB_ID, jobId,
         WorkerEnvConstants.WORKER_JOB_ATTEMPT, String.valueOf(attempt),
-        EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE, String.valueOf(featureFlags.useStreamCapableState()));
+        EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE, String.valueOf(featureFlags.useStreamCapableState()),
+        EnvVariableFeatureFlags.AUTO_DETECT_SCHEMA, String.valueOf(featureFlags.autoDetectSchema()));
   }
 
 }

airbyte-commons-worker/src/test/java/io/airbyte/workers/process/AirbyteIntegrationLauncherTest.java

@@ -53,7 +53,8 @@ class AirbyteIntegrationLauncherTest {
       WorkerEnvConstants.WORKER_CONNECTOR_IMAGE, FAKE_IMAGE,
       WorkerEnvConstants.WORKER_JOB_ID, JOB_ID,
       WorkerEnvConstants.WORKER_JOB_ATTEMPT, String.valueOf(JOB_ATTEMPT),
-      EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE, String.valueOf(new EnvVariableFeatureFlags().useStreamCapableState()));
+      EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE, String.valueOf(new EnvVariableFeatureFlags().useStreamCapableState()),
+      EnvVariableFeatureFlags.AUTO_DETECT_SCHEMA, String.valueOf(new EnvVariableFeatureFlags().autoDetectSchema()));
 
   private WorkerConfigs workerConfigs;
   @Mock

airbyte-commons/src/main/java/io/airbyte/commons/features/EnvVariableFeatureFlags.java

@@ -11,6 +11,7 @@
 public class EnvVariableFeatureFlags implements FeatureFlags {
 
   public static final String USE_STREAM_CAPABLE_STATE = "USE_STREAM_CAPABLE_STATE";
+  public static final String AUTO_DETECT_SCHEMA = "AUTO_DETECT_SCHEMA";
   public static final String LOG_CONNECTOR_MESSAGES = "LOG_CONNECTOR_MESSAGES";
   public static final String NEED_STATE_VALIDATION = "NEED_STATE_VALIDATION";
 
@@ -31,6 +32,11 @@ public boolean useStreamCapableState() {
     return getEnvOrDefault(USE_STREAM_CAPABLE_STATE, false, Boolean::parseBoolean);
   }
 
+  @Override
+  public boolean autoDetectSchema() {
+    return getEnvOrDefault(AUTO_DETECT_SCHEMA, false, Boolean::parseBoolean);
+  }
+
   @Override
   public boolean logConnectorMessages() {
     return getEnvOrDefault(LOG_CONNECTOR_MESSAGES, false, Boolean::parseBoolean);

airbyte-commons/src/main/java/io/airbyte/commons/features/FeatureFlags.java

@@ -16,6 +16,8 @@ public interface FeatureFlags {
 
   boolean useStreamCapableState();
 
+  boolean autoDetectSchema();
+
   boolean logConnectorMessages();
 
   boolean needStateValidation();

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

@@ -687,6 +687,8 @@ public interface Configs {
    */
   int getActivityNumberOfAttempt();
 
+  boolean getAutoDetectSchema();
+
   enum TrackingStrategy {
     SEGMENT,
     LOGGING

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

@@ -202,6 +202,7 @@ public class EnvConfigs implements Configs {
   private static final long DEFAULT_MAX_SYNC_WORKERS = 5;
   private static final long DEFAULT_MAX_NOTIFY_WORKERS = 5;
   private static final String DEFAULT_NETWORK = "host";
+  private static final String AUTO_DETECT_SCHEMA = "AUTO_DETECT_SCHEMA";
 
   public static final Map<String, Function<EnvConfigs, String>> JOB_SHARED_ENVS = Map.of(
       AIRBYTE_VERSION, (instance) -> instance.getAirbyteVersion().serialize(),
@@ -1050,6 +1051,11 @@ public int getWorkflowFailureRestartDelaySeconds() {
     return Integer.parseInt(getEnvOrDefault(WORKFLOW_FAILURE_RESTART_DELAY_SECONDS, String.valueOf(10 * 60)));
   }
 
+  @Override
+  public boolean getAutoDetectSchema() {
+    return getEnvOrDefault(AUTO_DETECT_SCHEMA, false);
+  }
+
   @Override
   public int getActivityNumberOfAttempt() {
     return Integer.parseInt(getEnvOrDefault(ACTIVITY_MAX_ATTEMPT, "5"));

airbyte-workers/src/main/java/io/airbyte/workers/config/ContainerOrchestratorConfigBeanFactory.java

@@ -76,6 +76,7 @@ public ContainerOrchestratorConfig kubernetesContainerOrchestratorConfig(
     environmentVariables.put(DD_DOGSTATSD_PORT_ENV_VAR, dataDogStatsdPort);
     environmentVariables.put(PUBLISH_METRICS_ENV_VAR, shouldPublishMetrics);
     environmentVariables.put(EnvVariableFeatureFlags.USE_STREAM_CAPABLE_STATE, Boolean.toString(featureFlags.useStreamCapableState()));
+    environmentVariables.put(EnvVariableFeatureFlags.AUTO_DETECT_SCHEMA, Boolean.toString(featureFlags.autoDetectSchema()));
     environmentVariables.put(JAVA_OPTS_ENV_VAR, containerOrchestratorJavaOpts);
 
     if (System.getenv(DD_ENV_ENV_VAR) != null) {