add nonBreakingChangePreference and notifySchemaChanges (#18636)

* add nonBreakingChangePreference and notifySchemaChanges

airbyte-api/src/main/openapi/config.yaml

@@ -3266,6 +3266,10 @@ components:
           format: uuid
         geography:
           $ref: "#/components/schemas/Geography"
+        notifySchemaChanges:
+          type: boolean
+        nonBreakingChangesPreference:
+          $ref: "#/components/schemas/NonBreakingChangesPreference"
     WebBackendConnectionCreate:
       type: object
       required:
@@ -3367,6 +3371,10 @@ components:
           format: uuid
         geography:
           $ref: "#/components/schemas/Geography"
+        notifySchemaChanges:
+          type: boolean
+        nonBreakingChangesPreference:
+          $ref: "#/components/schemas/NonBreakingChangesPreference"
     WebBackendConnectionUpdate:
       type: object
       description: Used to apply a patch-style update to a connection, which means that null properties remain unchanged
@@ -3411,6 +3419,10 @@ components:
           format: uuid
         geography:
           $ref: "#/components/schemas/Geography"
+        notifySchemaChanges:
+          type: boolean
+        nonBreakingChangesPreference:
+          $ref: "#/components/schemas/NonBreakingChangesPreference"
     ConnectionRead:
       type: object
       required:
@@ -3463,6 +3475,10 @@ components:
           $ref: "#/components/schemas/Geography"
         breakingChange:
           type: boolean
+        notifySchemaChanges:
+          type: boolean
+        nonBreakingChangesPreference:
+          $ref: "#/components/schemas/NonBreakingChangesPreference"
     SchemaChange:
       enum:
         - no_change
@@ -4713,6 +4729,8 @@ components:
         - destination
         - isSyncing
         - schemaChange
+        - notifySchemaChanges
+        - nonBreakingChangesPreference
       properties:
         connectionId:
           $ref: "#/components/schemas/ConnectionId"
@@ -4771,6 +4789,15 @@ components:
           $ref: "#/components/schemas/Geography"
         schemaChange:
           $ref: "#/components/schemas/SchemaChange"
+        notifySchemaChanges:
+          type: boolean
+        nonBreakingChangesPreference:
+          $ref: "#/components/schemas/NonBreakingChangesPreference"
+    NonBreakingChangesPreference:
+      enum:
+        - ignore
+        - disable
+      type: string
     WebBackendConnectionReadList:
       type: object
       required:

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

@@ -120,3 +120,10 @@ properties:
     "$ref": Geography.yaml
   breakingChange:
     type: boolean
+  notifySchemaChanges:
+    type: boolean
+  nonBreakingChangesPreference:
+    type: string
+    enum:
+      - ignore
+      - disable

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

@@ -32,6 +32,7 @@
 import io.airbyte.config.StandardSourceDefinition;
 import io.airbyte.config.StandardSourceDefinition.SourceType;
 import io.airbyte.config.StandardSync;
+import io.airbyte.config.StandardSync.NonBreakingChangesPreference;
 import io.airbyte.config.StandardSync.ScheduleType;
 import io.airbyte.config.StandardSync.Status;
 import io.airbyte.config.StandardWorkspace;
@@ -79,7 +80,10 @@ public static StandardSync buildStandardSync(final Record record, final List<UUI
             Jsons.deserialize(record.get(CONNECTION.RESOURCE_REQUIREMENTS).data(), ResourceRequirements.class))
         .withSourceCatalogId(record.get(CONNECTION.SOURCE_CATALOG_ID))
         .withBreakingChange(record.get(CONNECTION.BREAKING_CHANGE))
-        .withGeography(Enums.toEnum(record.get(CONNECTION.GEOGRAPHY, String.class), Geography.class).orElseThrow());
+        .withGeography(Enums.toEnum(record.get(CONNECTION.GEOGRAPHY, String.class), Geography.class).orElseThrow())
+        .withNonBreakingChangesPreference(
+            Enums.toEnum(record.get(CONNECTION.NON_BREAKING_CHANGE_PREFERENCE, String.class), NonBreakingChangesPreference.class).orElseThrow())
+        .withNotifySchemaChanges(record.get(CONNECTION.NOTIFY_SCHEMA_CHANGES));
   }
 
   public static StandardWorkspace buildStandardWorkspace(final Record record) {

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/MockData.java

@@ -30,6 +30,7 @@
 import io.airbyte.config.StandardSourceDefinition;
 import io.airbyte.config.StandardSourceDefinition.SourceType;
 import io.airbyte.config.StandardSync;
+import io.airbyte.config.StandardSync.NonBreakingChangesPreference;
 import io.airbyte.config.StandardSync.Status;
 import io.airbyte.config.StandardSyncOperation;
 import io.airbyte.config.StandardSyncOperation.OperatorType;
@@ -479,7 +480,9 @@ public static List<StandardSync> standardSyncs() {
         .withStatus(Status.ACTIVE)
         .withSchedule(schedule)
         .withGeography(Geography.AUTO)
-        .withBreakingChange(false);
+        .withBreakingChange(false)
+        .withNonBreakingChangesPreference(NonBreakingChangesPreference.IGNORE)
+        .withNotifySchemaChanges(true);
 
     final StandardSync standardSync2 = new StandardSync()
         .withOperationIds(Arrays.asList(OPERATION_ID_1, OPERATION_ID_2))
@@ -496,7 +499,9 @@ public static List<StandardSync> standardSyncs() {
         .withStatus(Status.ACTIVE)
         .withSchedule(schedule)
         .withGeography(Geography.AUTO)
-        .withBreakingChange(false);
+        .withBreakingChange(false)
+        .withNonBreakingChangesPreference(NonBreakingChangesPreference.IGNORE)
+        .withNotifySchemaChanges(true);
 
     final StandardSync standardSync3 = new StandardSync()
         .withOperationIds(Arrays.asList(OPERATION_ID_1, OPERATION_ID_2))
@@ -513,7 +518,9 @@ public static List<StandardSync> standardSyncs() {
         .withStatus(Status.ACTIVE)
         .withSchedule(schedule)
         .withGeography(Geography.AUTO)
-        .withBreakingChange(false);
+        .withBreakingChange(false)
+        .withNonBreakingChangesPreference(NonBreakingChangesPreference.IGNORE)
+        .withNotifySchemaChanges(true);
 
     final StandardSync standardSync4 = new StandardSync()
         .withOperationIds(Collections.emptyList())
@@ -530,7 +537,9 @@ public static List<StandardSync> standardSyncs() {
         .withStatus(Status.DEPRECATED)
         .withSchedule(schedule)
         .withGeography(Geography.AUTO)
-        .withBreakingChange(false);
+        .withBreakingChange(false)
+        .withNonBreakingChangesPreference(NonBreakingChangesPreference.IGNORE)
+        .withNotifySchemaChanges(true);
 
     final StandardSync standardSync5 = new StandardSync()
         .withOperationIds(Arrays.asList(OPERATION_ID_3))
@@ -547,7 +556,9 @@ public static List<StandardSync> standardSyncs() {
         .withStatus(Status.ACTIVE)
         .withSchedule(schedule)
         .withGeography(Geography.AUTO)
-        .withBreakingChange(false);
+        .withBreakingChange(false)
+        .withNonBreakingChangesPreference(NonBreakingChangesPreference.IGNORE)
+        .withNotifySchemaChanges(true);
 
     final StandardSync standardSync6 = new StandardSync()
         .withOperationIds(Arrays.asList())
@@ -564,7 +575,9 @@ public static List<StandardSync> standardSyncs() {
         .withStatus(Status.DEPRECATED)
         .withSchedule(schedule)
         .withGeography(Geography.AUTO)
-        .withBreakingChange(false);
+        .withBreakingChange(false)
+        .withNonBreakingChangesPreference(NonBreakingChangesPreference.IGNORE)
+        .withNotifySchemaChanges(true);
 
     return Arrays.asList(standardSync1, standardSync2, standardSync3, standardSync4, standardSync5, standardSync6);
   }

airbyte-server/src/main/java/io/airbyte/server/converters/ApiPojoConverters.java

@@ -14,6 +14,7 @@
 import io.airbyte.api.model.generated.Geography;
 import io.airbyte.api.model.generated.JobType;
 import io.airbyte.api.model.generated.JobTypeResourceLimit;
+import io.airbyte.api.model.generated.NonBreakingChangesPreference;
 import io.airbyte.api.model.generated.ResourceRequirements;
 import io.airbyte.commons.enums.Enums;
 import io.airbyte.config.BasicSchedule;
@@ -94,7 +95,9 @@ public static ConnectionRead internalToConnectionRead(final StandardSync standar
         .syncCatalog(CatalogConverter.toApi(standardSync.getCatalog()))
         .sourceCatalogId(standardSync.getSourceCatalogId())
         .breakingChange(standardSync.getBreakingChange())
-        .geography(Enums.convertTo(standardSync.getGeography(), Geography.class));
+        .geography(Enums.convertTo(standardSync.getGeography(), Geography.class))
+        .nonBreakingChangesPreference(Enums.convertTo(standardSync.getNonBreakingChangesPreference(), NonBreakingChangesPreference.class))
+        .notifySchemaChanges(standardSync.getNotifySchemaChanges());
 
     if (standardSync.getResourceRequirements() != null) {
       connectionRead.resourceRequirements(resourceRequirementsToApi(standardSync.getResourceRequirements()));

airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java

@@ -297,7 +297,9 @@ private static WebBackendConnectionRead getWebBackendConnectionRead(final Connec
         .destination(destination)
         .operations(operations.getOperations())
         .resourceRequirements(connectionRead.getResourceRequirements())
-        .geography(connectionRead.getGeography());
+        .geography(connectionRead.getGeography())
+        .notifySchemaChanges(connectionRead.getNotifySchemaChanges())
+        .nonBreakingChangesPreference(connectionRead.getNonBreakingChangesPreference());
   }
 
   // todo (cgardens) - This logic is a headache to follow it stems from the internal data model not
@@ -656,6 +658,8 @@ protected static ConnectionUpdate toConnectionPatch(final WebBackendConnectionUp
     connectionPatch.resourceRequirements(webBackendConnectionPatch.getResourceRequirements());
     connectionPatch.sourceCatalogId(webBackendConnectionPatch.getSourceCatalogId());
     connectionPatch.geography(webBackendConnectionPatch.getGeography());
+    connectionPatch.notifySchemaChanges(webBackendConnectionPatch.getNotifySchemaChanges());
+    connectionPatch.nonBreakingChangesPreference(webBackendConnectionPatch.getNonBreakingChangesPreference());
 
     connectionPatch.operationIds(finalOperationIds);
 

airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java

@@ -538,7 +538,7 @@ void testForConnectionCreateCompleteness() {
     final Set<String> handledMethods =
         Set.of("name", "namespaceDefinition", "namespaceFormat", "prefix", "sourceId", "destinationId", "operationIds",
             "addOperationIdsItem", "removeOperationIdsItem", "syncCatalog", "schedule", "scheduleType", "scheduleData",
-            "status", "resourceRequirements", "sourceCatalogId", "geography");
+            "status", "resourceRequirements", "sourceCatalogId", "geography", "nonBreakingChangesPreference", "notifySchemaChanges");
 
     final Set<String> methods = Arrays.stream(ConnectionCreate.class.getMethods())
         .filter(method -> method.getReturnType() == ConnectionCreate.class)
@@ -560,7 +560,7 @@ void testForConnectionPatchCompleteness() {
     final Set<String> handledMethods =
         Set.of("schedule", "connectionId", "syncCatalog", "namespaceDefinition", "namespaceFormat", "prefix", "status",
             "operationIds", "addOperationIdsItem", "removeOperationIdsItem", "resourceRequirements", "name",
-            "sourceCatalogId", "scheduleType", "scheduleData", "geography");
+            "sourceCatalogId", "scheduleType", "scheduleData", "geography", "notifySchemaChanges", "nonBreakingChangesPreference");
 
     final Set<String> methods = Arrays.stream(ConnectionUpdate.class.getMethods())
         .filter(method -> method.getReturnType() == ConnectionUpdate.class)

airbyte-webapp/src/test-utils/mock-data/mockConnection.json

@@ -346,5 +346,7 @@
   "latestSyncJobStatus": "succeeded",
   "isSyncing": false,
   "catalogId": "bf31d1df-d7ba-4bae-b1ec-dac617b4f70c",
-  "schemaChange": "no_change"
+  "schemaChange": "no_change",
+  "notifySchemaChanges": true,
+  "nonBreakingChangesPreference": "ignore"
 }

airbyte-webapp/src/test-utils/testutils.tsx

@@ -122,4 +122,6 @@ export const mockConnection: WebBackendConnectionRead = {
   catalogId: "",
   isSyncing: false,
   schemaChange: "no_change",
+  notifySchemaChanges: true,
+  nonBreakingChangesPreference: "ignore",
 };