diff --git a/airbyte-api/src/main/openapi/config.yaml b/airbyte-api/src/main/openapi/config.yaml
index 143987726ade..baa488de77df 100644
--- a/airbyte-api/src/main/openapi/config.yaml
+++ b/airbyte-api/src/main/openapi/config.yaml
@@ -4649,6 +4649,7 @@ components:
         - destination
         - status
         - isSyncing
+        - schemaChange
       properties:
         connectionId:
           $ref: "#/components/schemas/ConnectionId"
@@ -4674,6 +4675,8 @@ components:
           $ref: "#/components/schemas/JobStatus"
         isSyncing:
           type: boolean
+        schemaChange:
+          $ref: "#/components/schemas/SchemaChange"
     WebBackendConnectionRead:
       type: object
       required:
diff --git a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java
index f65c701d2b34..da59d93e9068 100644
--- a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java
+++ b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java
@@ -992,6 +992,22 @@ public Optional<ActorCatalogFetchEvent> getMostRecentActorCatalogFetchEventForSo
     return records.stream().findFirst().map(DbConverter::buildActorCatalogFetchEvent);
   }
 
+  public Map<UUID, ActorCatalogFetchEvent> getMostRecentActorCatalogFetchEventForSources(final List<UUID> sourceIds)
+      throws IOException {
+
+    return database.query(ctx -> ctx.fetch(
+        """
+        select actor_catalog_id, actor_id from
+          (select actor_catalog_id, actor_id, rank() over (partition by actor_id order by created_at desc) as creation_order_rank
+          from public.actor_catalog_fetch_event
+          ) table_with_rank
+        where creation_order_rank = 1;
+        """))
+        .stream().map(DbConverter::buildActorCatalogFetchEvent)
+        .collect(Collectors.toMap(record -> record.getActorId(),
+            record -> record));
+  }
+
   /**
    * Stores source catalog information.
    *
diff --git a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/DbConverter.java b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/DbConverter.java
index 6bdc881bc269..ea4548d9f78a 100644
--- a/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/DbConverter.java
+++ b/airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/DbConverter.java
@@ -198,6 +198,7 @@ public static ActorCatalog buildActorCatalog(final Record record) {
 
   public static ActorCatalogFetchEvent buildActorCatalogFetchEvent(final Record record) {
     return new ActorCatalogFetchEvent()
+        .withActorId(record.get(ACTOR_CATALOG_FETCH_EVENT.ACTOR_ID))
         .withActorCatalogId(record.get(ACTOR_CATALOG_FETCH_EVENT.ACTOR_CATALOG_ID));
   }
 
diff --git a/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/ConfigRepositoryE2EReadWriteTest.java b/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/ConfigRepositoryE2EReadWriteTest.java
index eea254e61d55..552355066b5e 100644
--- a/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/ConfigRepositoryE2EReadWriteTest.java
+++ b/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/ConfigRepositoryE2EReadWriteTest.java
@@ -515,17 +515,17 @@ void testGetGeographyForConnection() throws IOException {
   }
 
   @Test
-  void testGetMostRecentActorCatalogFetchEventForSources() throws SQLException, IOException, JsonValidationException {
+  void testGetMostRecentActorCatalogFetchEventForSource() throws SQLException, IOException, JsonValidationException {
     for (final ActorCatalog actorCatalog : MockData.actorCatalogs()) {
       configPersistence.writeConfig(ConfigSchema.ACTOR_CATALOG, actorCatalog.getId().toString(), actorCatalog);
     }
 
-    OffsetDateTime now = OffsetDateTime.now();
-    OffsetDateTime yesterday = now.minusDays(1l);
+    final OffsetDateTime now = OffsetDateTime.now();
+    final OffsetDateTime yesterday = now.minusDays(1l);
 
-    List<ActorCatalogFetchEvent> fetchEvents = MockData.actorCatalogFetchEventsSameSource();
-    ActorCatalogFetchEvent fetchEvent1 = fetchEvents.get(0);
-    ActorCatalogFetchEvent fetchEvent2 = fetchEvents.get(1);
+    final List<ActorCatalogFetchEvent> fetchEvents = MockData.actorCatalogFetchEventsSameSource();
+    final ActorCatalogFetchEvent fetchEvent1 = fetchEvents.get(0);
+    final ActorCatalogFetchEvent fetchEvent2 = fetchEvents.get(1);
 
     database.transaction(ctx -> {
       insertCatalogFetchEvent(
@@ -542,13 +542,37 @@ void testGetMostRecentActorCatalogFetchEventForSources() throws SQLException, IO
       return null;
     });
 
-    Optional<ActorCatalogFetchEvent> result =
+    final Optional<ActorCatalogFetchEvent> result =
         configRepository.getMostRecentActorCatalogFetchEventForSource(fetchEvent1.getActorId());
 
     assertEquals(fetchEvent2.getActorCatalogId(), result.get().getActorCatalogId());
   }
 
-  private void insertCatalogFetchEvent(DSLContext ctx, UUID sourceId, UUID catalogId, OffsetDateTime creationDate) {
+  @Test
+  void testGetMostRecentActorCatalogFetchEventForSources() throws SQLException, IOException, JsonValidationException {
+    for (final ActorCatalog actorCatalog : MockData.actorCatalogs()) {
+      configPersistence.writeConfig(ConfigSchema.ACTOR_CATALOG, actorCatalog.getId().toString(), actorCatalog);
+    }
+
+    database.transaction(ctx -> {
+      MockData.actorCatalogFetchEventsForAggregationTest().forEach(actorCatalogFetchEvent -> insertCatalogFetchEvent(
+          ctx,
+          actorCatalogFetchEvent.getActorCatalogFetchEvent().getActorId(),
+          actorCatalogFetchEvent.getActorCatalogFetchEvent().getActorCatalogId(),
+          actorCatalogFetchEvent.getCreatedAt()));
+
+      return null;
+    });
+
+    final Map<UUID, ActorCatalogFetchEvent> result =
+        configRepository.getMostRecentActorCatalogFetchEventForSources(List.of(MockData.SOURCE_ID_1,
+            MockData.SOURCE_ID_2));
+
+    assertEquals(MockData.ACTOR_CATALOG_ID_1, result.get(MockData.SOURCE_ID_1).getActorCatalogId());
+    assertEquals(MockData.ACTOR_CATALOG_ID_3, result.get(MockData.SOURCE_ID_2).getActorCatalogId());
+  }
+
+  private void insertCatalogFetchEvent(final DSLContext ctx, final UUID sourceId, final UUID catalogId, final OffsetDateTime creationDate) {
     ctx.insertInto(ACTOR_CATALOG_FETCH_EVENT)
         .columns(
             ACTOR_CATALOG_FETCH_EVENT.ID,
diff --git a/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/MockData.java b/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/MockData.java
index dddd705c30e3..2b6050b7d56a 100644
--- a/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/MockData.java
+++ b/airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/MockData.java
@@ -50,6 +50,7 @@
 import io.airbyte.protocol.models.SyncMode;
 import java.net.URI;
 import java.time.Instant;
+import java.time.OffsetDateTime;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -57,10 +58,11 @@
 import java.util.TreeMap;
 import java.util.UUID;
 import java.util.stream.Collectors;
+import lombok.Data;
 
 public class MockData {
 
-  private static final UUID WORKSPACE_ID_1 = UUID.randomUUID();
+  public static final UUID WORKSPACE_ID_1 = UUID.randomUUID();
   private static final UUID WORKSPACE_ID_2 = UUID.randomUUID();
   private static final UUID WORKSPACE_ID_3 = UUID.randomUUID();
   private static final UUID WORKSPACE_CUSTOMER_ID = UUID.randomUUID();
@@ -72,8 +74,8 @@ public class MockData {
   private static final UUID DESTINATION_DEFINITION_ID_2 = UUID.randomUUID();
   private static final UUID DESTINATION_DEFINITION_ID_3 = UUID.randomUUID();
   private static final UUID DESTINATION_DEFINITION_ID_4 = UUID.randomUUID();
-  private static final UUID SOURCE_ID_1 = UUID.randomUUID();
-  private static final UUID SOURCE_ID_2 = UUID.randomUUID();
+  public static final UUID SOURCE_ID_1 = UUID.randomUUID();
+  public static final UUID SOURCE_ID_2 = UUID.randomUUID();
   private static final UUID SOURCE_ID_3 = UUID.randomUUID();
   private static final UUID DESTINATION_ID_1 = UUID.randomUUID();
   private static final UUID DESTINATION_ID_2 = UUID.randomUUID();
@@ -91,11 +93,12 @@ public class MockData {
   private static final UUID SOURCE_OAUTH_PARAMETER_ID_2 = UUID.randomUUID();
   private static final UUID DESTINATION_OAUTH_PARAMETER_ID_1 = UUID.randomUUID();
   private static final UUID DESTINATION_OAUTH_PARAMETER_ID_2 = UUID.randomUUID();
-  private static final UUID ACTOR_CATALOG_ID_1 = UUID.randomUUID();
+  public static final UUID ACTOR_CATALOG_ID_1 = UUID.randomUUID();
   private static final UUID ACTOR_CATALOG_ID_2 = UUID.randomUUID();
-  private static final UUID ACTOR_CATALOG_ID_3 = UUID.randomUUID();
+  public static final UUID ACTOR_CATALOG_ID_3 = UUID.randomUUID();
   private static final UUID ACTOR_CATALOG_FETCH_EVENT_ID_1 = UUID.randomUUID();
   private static final UUID ACTOR_CATALOG_FETCH_EVENT_ID_2 = UUID.randomUUID();
+  private static final UUID ACTOR_CATALOG_FETCH_EVENT_ID_3 = UUID.randomUUID();
 
   public static final String MOCK_SERVICE_ACCOUNT_1 = "{\n"
       + "  \"type\" : \"service_account\",\n"
@@ -622,8 +625,8 @@ public static List<ActorCatalogFetchEvent> actorCatalogFetchEvents() {
         .withId(ACTOR_CATALOG_FETCH_EVENT_ID_2)
         .withActorCatalogId(ACTOR_CATALOG_ID_2)
         .withActorId(SOURCE_ID_2)
-        .withConfigHash("1394")
-        .withConnectorVersion("1.2.0");
+        .withConfigHash("1395")
+        .withConnectorVersion("1.42.0");
     return Arrays.asList(actorCatalogFetchEvent1, actorCatalogFetchEvent2);
   }
 
@@ -643,6 +646,42 @@ public static List<ActorCatalogFetchEvent> actorCatalogFetchEventsSameSource() {
     return Arrays.asList(actorCatalogFetchEvent1, actorCatalogFetchEvent2);
   }
 
+  @Data
+  public static class ActorCatalogFetchEventWithCreationDate {
+
+    private final ActorCatalogFetchEvent actorCatalogFetchEvent;
+    private final OffsetDateTime createdAt;
+
+  }
+
+  public static List<ActorCatalogFetchEventWithCreationDate> actorCatalogFetchEventsForAggregationTest() {
+    final OffsetDateTime now = OffsetDateTime.now();
+    final OffsetDateTime yesterday = OffsetDateTime.now().minusDays(1l);
+
+    final ActorCatalogFetchEvent actorCatalogFetchEvent1 = new ActorCatalogFetchEvent()
+        .withId(ACTOR_CATALOG_FETCH_EVENT_ID_1)
+        .withActorCatalogId(ACTOR_CATALOG_ID_1)
+        .withActorId(SOURCE_ID_1)
+        .withConfigHash("CONFIG_HASH")
+        .withConnectorVersion("1.0.0");
+    final ActorCatalogFetchEvent actorCatalogFetchEvent2 = new ActorCatalogFetchEvent()
+        .withId(ACTOR_CATALOG_FETCH_EVENT_ID_2)
+        .withActorCatalogId(ACTOR_CATALOG_ID_2)
+        .withActorId(SOURCE_ID_2)
+        .withConfigHash("1394")
+        .withConnectorVersion("1.2.0");
+    final ActorCatalogFetchEvent actorCatalogFetchEvent3 = new ActorCatalogFetchEvent()
+        .withId(ACTOR_CATALOG_FETCH_EVENT_ID_3)
+        .withActorCatalogId(ACTOR_CATALOG_ID_3)
+        .withActorId(SOURCE_ID_2)
+        .withConfigHash("1394")
+        .withConnectorVersion("1.2.0");
+    return Arrays.asList(
+        new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent1, now),
+        new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent2, yesterday),
+        new ActorCatalogFetchEventWithCreationDate(actorCatalogFetchEvent3, now));
+  }
+
   public static List<WorkspaceServiceAccount> workspaceServiceAccounts() {
     final WorkspaceServiceAccount workspaceServiceAccount = new WorkspaceServiceAccount()
         .withWorkspaceId(WORKSPACE_ID_1)
diff --git a/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java b/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java
index 60d980b09847..c18977a9e492 100644
--- a/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java
+++ b/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java
@@ -99,7 +99,8 @@ public ConnectionStateType getStateType(final ConnectionIdRequestBody connection
     return Enums.convertTo(stateHandler.getState(connectionIdRequestBody).getStateType(), ConnectionStateType.class);
   }
 
-  public WebBackendConnectionReadList webBackendListConnectionsForWorkspace(final WorkspaceIdRequestBody workspaceIdRequestBody) throws IOException {
+  public WebBackendConnectionReadList webBackendListConnectionsForWorkspace(final WorkspaceIdRequestBody workspaceIdRequestBody)
+      throws IOException, JsonValidationException, ConfigNotFoundException {
 
     // passing 'false' so that deleted connections are not included
     final List<StandardSync> standardSyncs =
@@ -113,6 +114,9 @@ public WebBackendConnectionReadList webBackendListConnectionsForWorkspace(final
     final Map<UUID, JobRead> runningJobByConnectionId =
         getRunningJobByConnectionId(standardSyncs.stream().map(StandardSync::getConnectionId).toList());
 
+    final Map<UUID, ActorCatalogFetchEvent> newestFetchEventsByActorId =
+        configRepository.getMostRecentActorCatalogFetchEventForSources(new ArrayList<>());
+
     final List<WebBackendConnectionListItem> connectionItems = Lists.newArrayList();
 
     for (final StandardSync standardSync : standardSyncs) {
@@ -122,7 +126,8 @@ public WebBackendConnectionReadList webBackendListConnectionsForWorkspace(final
               sourceReadById,
               destinationReadById,
               latestJobByConnectionId,
-              runningJobByConnectionId));
+              runningJobByConnectionId,
+              Optional.ofNullable(newestFetchEventsByActorId.get(standardSync.getSourceId()))));
     }
 
     return new WebBackendConnectionReadList().connections(connectionItems);
@@ -175,51 +180,33 @@ private WebBackendConnectionRead buildWebBackendConnectionRead(final ConnectionR
       webBackendConnectionRead.setLatestSyncJobStatus(job.getStatus());
     });
 
-    SchemaChange schemaChange = getSchemaChange(connectionRead, currentSourceCatalogId);
+    final Optional<ActorCatalogFetchEvent> mostRecentFetchEvent =
+        configRepository.getMostRecentActorCatalogFetchEventForSource(connectionRead.getSourceId());
+
+    final SchemaChange schemaChange = getSchemaChange(connectionRead, currentSourceCatalogId, mostRecentFetchEvent);
 
     webBackendConnectionRead.setSchemaChange(schemaChange);
 
     return webBackendConnectionRead;
   }
 
-  /*
-   * A breakingChange boolean is stored on the connectionRead object and corresponds to the boolean
-   * breakingChange field on the connection table. If there is not a breaking change, we still have to
-   * check whether there is a non-breaking schema change by fetching the most recent
-   * ActorCatalogFetchEvent. A new ActorCatalogFetchEvent is stored each time there is a source schema
-   * refresh, so if the most recent ActorCatalogFetchEvent has a different actor catalog than the
-   * existing actor catalog, there is a schema change.
-   */
-  private SchemaChange getSchemaChange(ConnectionRead connectionRead, Optional<UUID> currentSourceCatalogId) throws IOException {
-    SchemaChange schemaChange = SchemaChange.NO_CHANGE;
-
-    if (connectionRead.getBreakingChange()) {
-      schemaChange = SchemaChange.BREAKING;
-    } else if (currentSourceCatalogId.isPresent()) {
-      final Optional<ActorCatalogFetchEvent> mostRecentFetchEvent =
-          configRepository.getMostRecentActorCatalogFetchEventForSource(connectionRead.getSourceId());
-
-      if (mostRecentFetchEvent.isPresent()) {
-        if (!mostRecentFetchEvent.get().getActorCatalogId().equals(currentSourceCatalogId.get())) {
-          schemaChange = SchemaChange.NON_BREAKING;
-        }
-      }
-    }
-
-    return schemaChange;
-  }
-
   private WebBackendConnectionListItem buildWebBackendConnectionListItem(
                                                                          final StandardSync standardSync,
                                                                          final Map<UUID, SourceRead> sourceReadById,
                                                                          final Map<UUID, DestinationRead> destinationReadById,
                                                                          final Map<UUID, JobRead> latestJobByConnectionId,
-                                                                         final Map<UUID, JobRead> runningJobByConnectionId) {
+                                                                         final Map<UUID, JobRead> runningJobByConnectionId,
+                                                                         final Optional<ActorCatalogFetchEvent> latestFetchEvent)
+      throws JsonValidationException, ConfigNotFoundException, IOException {
 
     final SourceRead source = sourceReadById.get(standardSync.getSourceId());
     final DestinationRead destination = destinationReadById.get(standardSync.getDestinationId());
     final Optional<JobRead> latestSyncJob = Optional.ofNullable(latestJobByConnectionId.get(standardSync.getConnectionId()));
     final Optional<JobRead> latestRunningSyncJob = Optional.ofNullable(runningJobByConnectionId.get(standardSync.getConnectionId()));
+    final ConnectionRead connectionRead = connectionsHandler.getConnection(standardSync.getConnectionId());
+    final Optional<UUID> currentCatalogId = connectionRead == null ? Optional.empty() : Optional.ofNullable(connectionRead.getSourceCatalogId());
+
+    final SchemaChange schemaChange = getSchemaChange(connectionRead, currentCatalogId, latestFetchEvent);
 
     final WebBackendConnectionListItem listItem = new WebBackendConnectionListItem()
         .connectionId(standardSync.getConnectionId())
@@ -230,7 +217,8 @@ private WebBackendConnectionListItem buildWebBackendConnectionListItem(
         .scheduleType(ApiPojoConverters.toApiConnectionScheduleType(standardSync))
         .scheduleData(ApiPojoConverters.toApiConnectionScheduleData(standardSync))
         .source(source)
-        .destination(destination);
+        .destination(destination)
+        .schemaChange(schemaChange);
 
     listItem.setIsSyncing(latestRunningSyncJob.isPresent());
 
@@ -242,6 +230,34 @@ private WebBackendConnectionListItem buildWebBackendConnectionListItem(
     return listItem;
   }
 
+  /*
+   * A breakingChange boolean is stored on the connectionRead object and corresponds to the boolean
+   * breakingChange field on the connection table. If there is not a breaking change, we still have to
+   * check whether there is a non-breaking schema change by fetching the most recent
+   * ActorCatalogFetchEvent. A new ActorCatalogFetchEvent is stored each time there is a source schema
+   * refresh, so if the most recent ActorCatalogFetchEvent has a different actor catalog than the
+   * existing actor catalog, there is a schema change.
+   */
+  @VisibleForTesting
+  SchemaChange getSchemaChange(
+                               final ConnectionRead connectionRead,
+                               final Optional<UUID> currentSourceCatalogId,
+                               final Optional<ActorCatalogFetchEvent> mostRecentFetchEvent) {
+    if (connectionRead == null || currentSourceCatalogId.isEmpty()) {
+      return SchemaChange.NO_CHANGE;
+    }
+
+    if (connectionRead.getBreakingChange() != null && connectionRead.getBreakingChange()) {
+      return SchemaChange.BREAKING;
+    }
+
+    if (mostRecentFetchEvent.isPresent() && !mostRecentFetchEvent.map(ActorCatalogFetchEvent::getActorCatalogId).equals(currentSourceCatalogId)) {
+      return SchemaChange.NON_BREAKING;
+    }
+
+    return SchemaChange.NO_CHANGE;
+  }
+
   private SourceRead getSourceRead(final UUID sourceId) throws JsonValidationException, IOException, ConfigNotFoundException {
     final SourceIdRequestBody sourceIdRequestBody = new SourceIdRequestBody().sourceId(sourceId);
     return sourceHandler.getSource(sourceIdRequestBody);
diff --git a/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java b/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java
index 9efd1ca4d6e8..646c09ad0b6b 100644
--- a/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java
+++ b/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java
@@ -251,7 +251,8 @@ void setup() throws IOException, JsonValidationException, ConfigNotFoundExceptio
         destinationRead,
         false,
         jobRead.getJob().getCreatedAt(),
-        jobRead.getJob().getStatus());
+        jobRead.getJob().getStatus(),
+        SchemaChange.NO_CHANGE);
 
     expected = expectedWebBackendConnectionReadObject(connectionRead, sourceRead, destinationRead, operationReadList, SchemaChange.NO_CHANGE, now,
         connectionRead.getSyncCatalog(), connectionRead.getSourceCatalogId());
@@ -345,7 +346,7 @@ void testGetWorkspaceStateEmpty() throws IOException {
   }
 
   @Test
-  void testWebBackendListConnectionsForWorkspace() throws IOException {
+  void testWebBackendListConnectionsForWorkspace() throws IOException, JsonValidationException, ConfigNotFoundException {
     final WorkspaceIdRequestBody workspaceIdRequestBody = new WorkspaceIdRequestBody();
     workspaceIdRequestBody.setWorkspaceId(sourceRead.getWorkspaceId());
 
@@ -371,9 +372,9 @@ void testWebBackendGetConnection() throws ConfigNotFoundException, IOException,
     when(connectionsHandler.getConnection(connectionRead.getConnectionId())).thenReturn(connectionRead);
     when(operationsHandler.listOperationsForConnection(connectionIdRequestBody)).thenReturn(operationReadList);
 
-    final WebBackendConnectionRead WebBackendConnectionRead = wbHandler.webBackendGetConnection(webBackendConnectionRequestBody);
+    final WebBackendConnectionRead webBackendConnectionRead = wbHandler.webBackendGetConnection(webBackendConnectionRequestBody);
 
-    assertEquals(expected, WebBackendConnectionRead);
+    assertEquals(expected, webBackendConnectionRead);
 
     // make sure the icons were loaded into actual svg content
     assertTrue(expected.getSource().getIcon().startsWith(SVG));
@@ -404,8 +405,9 @@ void testWebBackendGetConnectionWithDiscoveryAndNewSchema() throws ConfigNotFoun
       IOException, JsonValidationException {
     when(connectionsHandler.getDiff(any(), any(),
         any())).thenReturn(expectedWithNewSchema.getCatalogDiff());
+    final UUID newCatalogId = UUID.randomUUID();
     when(configRepository.getMostRecentActorCatalogFetchEventForSource(any()))
-        .thenReturn(Optional.of(new ActorCatalogFetchEvent().withActorCatalogId(UUID.randomUUID())));
+        .thenReturn(Optional.of(new ActorCatalogFetchEvent().withActorCatalogId(newCatalogId)));
     when(configRepository.getActorCatalogById(any())).thenReturn(new ActorCatalog().withId(UUID.randomUUID()));
     final WebBackendConnectionRead result = testWebBackendGetConnection(true, connectionRead,
         operationReadList);
@@ -1134,4 +1136,38 @@ void testGetStreamsToReset() {
             streamDescriptor -> streamDescriptor.getName() == "updated_stream"));
   }
 
+  @Test
+  void testGetSchemaChangeNoChange() {
+    final ConnectionRead connectionReadNotBreaking = new ConnectionRead().breakingChange(false);
+
+    assertEquals(SchemaChange.NO_CHANGE, wbHandler.getSchemaChange(null, Optional.of(UUID.randomUUID()), Optional.of(new ActorCatalogFetchEvent())));
+    assertEquals(SchemaChange.NO_CHANGE,
+        wbHandler.getSchemaChange(connectionReadNotBreaking, Optional.empty(), Optional.of(new ActorCatalogFetchEvent())));
+
+    final UUID catalogId = UUID.randomUUID();
+
+    assertEquals(SchemaChange.NO_CHANGE, wbHandler.getSchemaChange(connectionReadNotBreaking, Optional.of(catalogId),
+        Optional.of(new ActorCatalogFetchEvent().withActorCatalogId(catalogId))));
+  }
+
+  @Test
+  void testGetSchemaChangeBreaking() {
+    final UUID sourceId = UUID.randomUUID();
+    final ConnectionRead connectionReadWithSourceId = new ConnectionRead().sourceCatalogId(UUID.randomUUID()).sourceId(sourceId).breakingChange(true);
+
+    assertEquals(SchemaChange.BREAKING, wbHandler.getSchemaChange(connectionReadWithSourceId,
+        Optional.of(UUID.randomUUID()), Optional.empty()));
+  }
+
+  @Test
+  void testGetSchemaChangeNotBreaking() {
+    final UUID catalogId = UUID.randomUUID();
+    final UUID differentCatalogId = UUID.randomUUID();
+    final ConnectionRead connectionReadWithSourceId =
+        new ConnectionRead().breakingChange(false);
+
+    assertEquals(SchemaChange.NON_BREAKING, wbHandler.getSchemaChange(connectionReadWithSourceId,
+        Optional.of(catalogId), Optional.of(new ActorCatalogFetchEvent().withActorCatalogId(differentCatalogId))));
+  }
+
 }
diff --git a/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java b/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java
index 89984a323944..ab742a079b36 100644
--- a/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java
+++ b/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java
@@ -21,6 +21,7 @@
 import io.airbyte.api.model.generated.Geography;
 import io.airbyte.api.model.generated.JobStatus;
 import io.airbyte.api.model.generated.ResourceRequirements;
+import io.airbyte.api.model.generated.SchemaChange;
 import io.airbyte.api.model.generated.SourceRead;
 import io.airbyte.api.model.generated.SyncMode;
 import io.airbyte.api.model.generated.WebBackendConnectionListItem;
@@ -243,7 +244,8 @@ public static WebBackendConnectionListItem generateExpectedWebBackendConnectionL
                                                                                           final DestinationRead destination,
                                                                                           final boolean isSyncing,
                                                                                           final Long latestSyncJobCreatedAt,
-                                                                                          final JobStatus latestSynJobStatus) {
+                                                                                          final JobStatus latestSynJobStatus,
+                                                                                          final SchemaChange schemaChange) {
 
     final WebBackendConnectionListItem connectionListItem = new WebBackendConnectionListItem()
         .connectionId(standardSync.getConnectionId())
@@ -257,7 +259,8 @@ public static WebBackendConnectionListItem generateExpectedWebBackendConnectionL
         .latestSyncJobCreatedAt(latestSyncJobCreatedAt)
         .latestSyncJobStatus(latestSynJobStatus)
         .scheduleType(ApiPojoConverters.toApiConnectionScheduleType(standardSync))
-        .scheduleData(ApiPojoConverters.toApiConnectionScheduleData(standardSync));
+        .scheduleData(ApiPojoConverters.toApiConnectionScheduleData(standardSync))
+        .schemaChange(schemaChange);
 
     return connectionListItem;
   }
diff --git a/docs/reference/api/generated-api-html/index.html b/docs/reference/api/generated-api-html/index.html
index 97d50bbbf698..af3d93edd239 100644
--- a/docs/reference/api/generated-api-html/index.html
+++ b/docs/reference/api/generated-api-html/index.html
@@ -8797,7 +8797,6 @@ <h3 class="field-label">Example data</h3>
   "connections" : [ {
     "sourceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
     "latestSyncJobCreatedAt" : 0,
-    "name" : "name",
     "destination" : {
       "connectionConfiguration" : {
         "user" : "charles"
@@ -8809,7 +8808,6 @@ <h3 class="field-label">Example data</h3>
       "destinationId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
       "workspaceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91"
     },
-    "connectionId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
     "isSyncing" : true,
     "source" : {
       "sourceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
@@ -8823,6 +8821,8 @@ <h3 class="field-label">Example data</h3>
       "workspaceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91"
     },
     "destinationId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
+    "name" : "name",
+    "connectionId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
     "scheduleData" : {
       "cron" : {
         "cronExpression" : "cronExpression",
@@ -8836,7 +8836,6 @@ <h3 class="field-label">Example data</h3>
   }, {
     "sourceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
     "latestSyncJobCreatedAt" : 0,
-    "name" : "name",
     "destination" : {
       "connectionConfiguration" : {
         "user" : "charles"
@@ -8848,7 +8847,6 @@ <h3 class="field-label">Example data</h3>
       "destinationId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
       "workspaceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91"
     },
-    "connectionId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
     "isSyncing" : true,
     "source" : {
       "sourceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
@@ -8862,6 +8860,8 @@ <h3 class="field-label">Example data</h3>
       "workspaceId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91"
     },
     "destinationId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
+    "name" : "name",
+    "connectionId" : "046b6c7f-0b8a-43b9-b35d-6489e6daee91",
     "scheduleData" : {
       "cron" : {
         "cronExpression" : "cronExpression",
@@ -11409,6 +11409,7 @@ <h3><a name="WebBackendConnectionListItem"><code>WebBackendConnectionListItem</c
 <div class="param">latestSyncJobCreatedAt (optional)</div><div class="param-desc"><span class="param-type"><a href="#long">Long</a></span> epoch time of the latest sync job. null if no sync job has taken place. format: int64</div>
 <div class="param">latestSyncJobStatus (optional)</div><div class="param-desc"><span class="param-type"><a href="#JobStatus">JobStatus</a></span>  </div>
 <div class="param">isSyncing </div><div class="param-desc"><span class="param-type"><a href="#boolean">Boolean</a></span>  </div>
+<div class="param">schemaChange </div><div class="param-desc"><span class="param-type"><a href="#SchemaChange">SchemaChange</a></span>  </div>
     </div>  <!-- field-items -->
   </div>
   <div class="model">