Fix connection read object building (#13568)

Make sure the sourceCatalogId field is set with the value stored in the
database.

Author: malikdiarra

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

@@ -130,7 +130,8 @@ public static ConnectionRead internalToConnectionRead(final StandardSync standar
         .namespaceDefinition(Enums.convertTo(standardSync.getNamespaceDefinition(), io.airbyte.api.model.generated.NamespaceDefinitionType.class))
         .namespaceFormat(standardSync.getNamespaceFormat())
         .prefix(standardSync.getPrefix())
-        .syncCatalog(CatalogConverter.toApi(standardSync.getCatalog()));
+        .syncCatalog(CatalogConverter.toApi(standardSync.getCatalog()))
+        .sourceCatalogId(standardSync.getSourceCatalogId());
 
     if (standardSync.getResourceRequirements() != null) {
       connectionRead.resourceRequirements(resourceRequirementsToApi(standardSync.getResourceRequirements()));

airbyte-server/src/main/java/io/airbyte/server/handlers/helpers/ConnectionMatcher.java

@@ -40,6 +40,7 @@ public ConnectionRead match(final ConnectionRead query) {
     fromSearch.resourceRequirements(query.getResourceRequirements());
     fromSearch.syncCatalog(query.getSyncCatalog());
     fromSearch.operationIds(query.getOperationIds());
+    fromSearch.sourceCatalogId(query.getSourceCatalogId());
 
     return fromSearch;
   }

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

@@ -115,7 +115,8 @@ void setUp() throws IOException, JsonValidationException, ConfigNotFoundExceptio
         .withOperationIds(List.of(operationId))
         .withManual(false)
         .withSchedule(ConnectionHelpers.generateBasicSchedule())
-        .withResourceRequirements(ConnectionHelpers.TESTING_RESOURCE_REQUIREMENTS);
+        .withResourceRequirements(ConnectionHelpers.TESTING_RESOURCE_REQUIREMENTS)
+        .withSourceCatalogId(UUID.randomUUID());
     standardSyncDeleted = new StandardSync()
         .withConnectionId(connectionId)
         .withName("presto to hudi2")
@@ -194,7 +195,8 @@ void testCreateConnection() throws JsonValidationException, ConfigNotFoundExcept
               .cpuRequest(standardSync.getResourceRequirements().getCpuRequest())
               .cpuLimit(standardSync.getResourceRequirements().getCpuLimit())
               .memoryRequest(standardSync.getResourceRequirements().getMemoryRequest())
-              .memoryLimit(standardSync.getResourceRequirements().getMemoryLimit()));
+              .memoryLimit(standardSync.getResourceRequirements().getMemoryLimit()))
+          .sourceCatalogId(standardSync.getSourceCatalogId());
 
       final ConnectionRead actualConnectionRead = connectionsHandler.createConnection(connectionCreate);
 
@@ -343,7 +345,8 @@ void testUpdateConnection() throws JsonValidationException, ConfigNotFoundExcept
           standardSync.getConnectionId(),
           standardSync.getSourceId(),
           standardSync.getDestinationId(),
-          standardSync.getOperationIds())
+          standardSync.getOperationIds(),
+          newSourceCatalogId)
           .schedule(null)
           .syncCatalog(catalog)
           .status(ConnectionStatus.INACTIVE);

airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java

@@ -99,7 +99,8 @@ public static Schedule generateBasicSchedule() {
   public static ConnectionRead generateExpectedConnectionRead(final UUID connectionId,
                                                               final UUID sourceId,
                                                               final UUID destinationId,
-                                                              final List<UUID> operationIds) {
+                                                              final List<UUID> operationIds,
+                                                              final UUID sourceCatalogId) {
 
     return new ConnectionRead()
         .connectionId(connectionId)
@@ -117,15 +118,17 @@ public static ConnectionRead generateExpectedConnectionRead(final UUID connectio
             .cpuRequest(TESTING_RESOURCE_REQUIREMENTS.getCpuRequest())
             .cpuLimit(TESTING_RESOURCE_REQUIREMENTS.getCpuLimit())
             .memoryRequest(TESTING_RESOURCE_REQUIREMENTS.getMemoryRequest())
-            .memoryLimit(TESTING_RESOURCE_REQUIREMENTS.getMemoryLimit()));
+            .memoryLimit(TESTING_RESOURCE_REQUIREMENTS.getMemoryLimit()))
+        .sourceCatalogId(sourceCatalogId);
   }
 
   public static ConnectionRead generateExpectedConnectionRead(final StandardSync standardSync) {
     final ConnectionRead connectionRead = generateExpectedConnectionRead(
         standardSync.getConnectionId(),
         standardSync.getSourceId(),
         standardSync.getDestinationId(),
-        standardSync.getOperationIds());
+        standardSync.getOperationIds(),
+        standardSync.getSourceCatalogId());
 
     if (standardSync.getSchedule() == null) {
       connectionRead.schedule(null);
@@ -147,7 +150,8 @@ public static ConnectionRead connectionReadFromStandardSync(final StandardSync s
         .operationIds(standardSync.getOperationIds())
         .name(standardSync.getName())
         .namespaceFormat(standardSync.getNamespaceFormat())
-        .prefix(standardSync.getPrefix());
+        .prefix(standardSync.getPrefix())
+        .sourceCatalogId(standardSync.getSourceCatalogId());
 
     if (standardSync.getNamespaceDefinition() != null) {
       connectionRead