FIX#5916: Only admin user retrieves connection params when external Secrets Manager is configured (#6228)

* Implementation for stop sending connection credentials when user is BOT

* Change way we add the connection to the service in the Secret Manager

* Services connection is not required as we want to stop returning it when SM is configured

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/ServiceEntityResource.java

@@ -0,0 +1,48 @@
+package org.openmetadata.catalog.resources.services;
+
+import java.util.Collections;
+import java.util.Optional;
+import javax.ws.rs.core.SecurityContext;
+import org.openmetadata.catalog.ServiceConnectionEntityInterface;
+import org.openmetadata.catalog.ServiceEntityInterface;
+import org.openmetadata.catalog.jdbi3.ServiceRepository;
+import org.openmetadata.catalog.resources.EntityResource;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.AuthorizationException;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.util.ResultList;
+
+public abstract class ServiceEntityResource<
+        T extends ServiceEntityInterface, R extends ServiceRepository<T, S>, S extends ServiceConnectionEntityInterface>
+    extends EntityResource<T, R> {
+
+  private final SecretsManager secretsManager;
+
+  protected ServiceEntityResource(
+      Class<T> entityClass, R serviceRepository, Authorizer authorizer, SecretsManager secretsManager) {
+    super(entityClass, serviceRepository, authorizer);
+    this.secretsManager = secretsManager;
+  }
+
+  protected T decryptOrNullify(SecurityContext securityContext, T service) {
+    try {
+      authorizer.authorizeAdmin(securityContext, secretsManager.isLocal());
+    } catch (AuthorizationException e) {
+      return nullifyConnection(service);
+    }
+    secretsManager.encryptOrDecryptServiceConnection(
+        service.getConnection(), extractServiceType(service), service.getName(), false);
+    return service;
+  }
+
+  protected ResultList<T> decryptOrNullify(SecurityContext securityContext, ResultList<T> services) {
+    Optional.ofNullable(services.getData())
+        .orElse(Collections.emptyList())
+        .forEach(mlModelService -> decryptOrNullify(securityContext, mlModelService));
+    return services;
+  }
+
+  protected abstract T nullifyConnection(T service);
+
+  protected abstract String extractServiceType(T service);
+}

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/dashboard/DashboardServiceResource.java

@@ -22,9 +22,7 @@ import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -51,10 +49,10 @@ import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.DashboardServiceRepository;
 import org.openmetadata.catalog.jdbi3.ListFilter;
 import org.openmetadata.catalog.resources.Collection;
-import org.openmetadata.catalog.resources.EntityResource;
+import org.openmetadata.catalog.resources.services.ServiceEntityResource;
 import org.openmetadata.catalog.secrets.SecretsManager;
-import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.DashboardConnection;
 import org.openmetadata.catalog.type.EntityHistory;
 import org.openmetadata.catalog.type.Include;
 import org.openmetadata.catalog.util.JsonUtils;
@@ -66,12 +64,11 @@ import org.openmetadata.catalog.util.ResultList;
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
 @Collection(name = "dashboardServices")
-public class DashboardServiceResource extends EntityResource<DashboardService, DashboardServiceRepository> {
+public class DashboardServiceResource
+    extends ServiceEntityResource<DashboardService, DashboardServiceRepository, DashboardConnection> {
   public static final String COLLECTION_PATH = "v1/services/dashboardServices";
   static final String FIELDS = FIELD_OWNER;
 
-  private final SecretsManager secretsManager;
-
   @Override
   public DashboardService addHref(UriInfo uriInfo, DashboardService service) {
     service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
@@ -80,8 +77,7 @@ public class DashboardServiceResource extends EntityResource<DashboardService, D
   }
 
   public DashboardServiceResource(CollectionDAO dao, Authorizer authorizer, SecretsManager secretsManager) {
-    super(DashboardService.class, new DashboardServiceRepository(dao, secretsManager), authorizer);
+    super(DashboardService.class, new DashboardServiceRepository(dao, secretsManager), authorizer, secretsManager);
-    this.secretsManager = secretsManager;
   }
 
   public static class DashboardServiceList extends ResultList<DashboardService> {
@@ -356,22 +352,13 @@ public class DashboardServiceResource extends EntityResource<DashboardService, D
         .withConnection(create.getConnection());
   }
 
-  private ResultList<DashboardService> decryptOrNullify(
+  @Override
-      SecurityContext securityContext, ResultList<DashboardService> dashboardServices) {
+  protected DashboardService nullifyConnection(DashboardService service) {
-    Optional.ofNullable(dashboardServices.getData())
+    return service.withConnection(null);
-        .orElse(Collections.emptyList())
-        .forEach(dashboardService -> decryptOrNullify(securityContext, dashboardService));
-    return dashboardServices;
   }
 
-  private DashboardService decryptOrNullify(SecurityContext securityContext, DashboardService dashboardService) {
+  @Override
-    try {
+  protected String extractServiceType(DashboardService service) {
-      authorizer.authorizeAdmin(securityContext, true);
+    return service.getServiceType().value();
-    } catch (AuthorizationException e) {
-      return dashboardService.withConnection(null);
-    }
-    secretsManager.encryptOrDecryptServiceConnection(
-        dashboardService.getConnection(), dashboardService.getServiceType().value(), dashboardService.getName(), false);
-    return dashboardService;
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/database/DatabaseServiceResource.java

@@ -20,9 +20,7 @@ import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -45,14 +43,14 @@ import javax.ws.rs.core.UriInfo;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.api.services.CreateDatabaseService;
+import org.openmetadata.catalog.api.services.DatabaseConnection;
 import org.openmetadata.catalog.entity.services.DatabaseService;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.DatabaseServiceRepository;
 import org.openmetadata.catalog.jdbi3.ListFilter;
 import org.openmetadata.catalog.resources.Collection;
-import org.openmetadata.catalog.resources.EntityResource;
+import org.openmetadata.catalog.resources.services.ServiceEntityResource;
 import org.openmetadata.catalog.secrets.SecretsManager;
-import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.type.EntityHistory;
 import org.openmetadata.catalog.type.Include;
@@ -67,12 +65,11 @@ import org.openmetadata.catalog.util.ResultList;
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
 @Collection(name = "databaseServices")
-public class DatabaseServiceResource extends EntityResource<DatabaseService, DatabaseServiceRepository> {
+public class DatabaseServiceResource
+    extends ServiceEntityResource<DatabaseService, DatabaseServiceRepository, DatabaseConnection> {
   public static final String COLLECTION_PATH = "v1/services/databaseServices/";
   static final String FIELDS = "pipelines,owner";
 
-  private final SecretsManager secretsManager;
-
   @Override
   public DatabaseService addHref(UriInfo uriInfo, DatabaseService service) {
     service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
@@ -82,8 +79,7 @@ public class DatabaseServiceResource extends EntityResource<DatabaseService, Dat
   }
 
   public DatabaseServiceResource(CollectionDAO dao, Authorizer authorizer, SecretsManager secretsManager) {
-    super(DatabaseService.class, new DatabaseServiceRepository(dao, secretsManager), authorizer);
+    super(DatabaseService.class, new DatabaseServiceRepository(dao, secretsManager), authorizer, secretsManager);
-    this.secretsManager = secretsManager;
   }
 
   public static class DatabaseServiceList extends ResultList<DatabaseService> {
@@ -362,22 +358,13 @@ public class DatabaseServiceResource extends EntityResource<DatabaseService, Dat
         .withConnection(create.getConnection());
   }
 
-  private ResultList<DatabaseService> decryptOrNullify(
+  @Override
-      SecurityContext securityContext, ResultList<DatabaseService> databaseServices) {
+  protected DatabaseService nullifyConnection(DatabaseService service) {
-    Optional.ofNullable(databaseServices.getData())
+    return service.withConnection(null);
-        .orElse(Collections.emptyList())
-        .forEach(databaseService -> decryptOrNullify(securityContext, databaseService));
-    return databaseServices;
   }
 
-  private DatabaseService decryptOrNullify(SecurityContext securityContext, DatabaseService databaseService) {
+  @Override
-    try {
+  protected String extractServiceType(DatabaseService service) {
-      authorizer.authorizeAdmin(securityContext, true);
+    return service.getServiceType().value();
-    } catch (AuthorizationException e) {
-      return databaseService.withConnection(null);
-    }
-    secretsManager.encryptOrDecryptServiceConnection(
-        databaseService.getConnection(), databaseService.getServiceType().value(), databaseService.getName(), false);
-    return databaseService;
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/messaging/MessagingServiceResource.java

@@ -22,9 +22,7 @@ import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -51,12 +49,12 @@ import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.ListFilter;
 import org.openmetadata.catalog.jdbi3.MessagingServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
-import org.openmetadata.catalog.resources.EntityResource;
+import org.openmetadata.catalog.resources.services.ServiceEntityResource;
 import org.openmetadata.catalog.secrets.SecretsManager;
-import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.type.EntityHistory;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.type.MessagingConnection;
 import org.openmetadata.catalog.util.JsonUtils;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.ResultList;
@@ -66,13 +64,12 @@ import org.openmetadata.catalog.util.ResultList;
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
 @Collection(name = "messagingServices")
-public class MessagingServiceResource extends EntityResource<MessagingService, MessagingServiceRepository> {
+public class MessagingServiceResource
+    extends ServiceEntityResource<MessagingService, MessagingServiceRepository, MessagingConnection> {
   public static final String COLLECTION_PATH = "v1/services/messagingServices/";
 
   public static final String FIELDS = FIELD_OWNER;
 
-  private final SecretsManager secretsManager;
-
   @Override
   public MessagingService addHref(UriInfo uriInfo, MessagingService service) {
     service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
@@ -81,8 +78,7 @@ public class MessagingServiceResource extends EntityResource<MessagingService, M
   }
 
   public MessagingServiceResource(CollectionDAO dao, Authorizer authorizer, SecretsManager secretsManager) {
-    super(MessagingService.class, new MessagingServiceRepository(dao, secretsManager), authorizer);
+    super(MessagingService.class, new MessagingServiceRepository(dao, secretsManager), authorizer, secretsManager);
-    this.secretsManager = secretsManager;
   }
 
   public static class MessagingServiceList extends ResultList<MessagingService> {
@@ -296,7 +292,9 @@ public class MessagingServiceResource extends EntityResource<MessagingService, M
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateMessagingService create)
       throws IOException {
     MessagingService service = getService(create, securityContext.getUserPrincipal().getName());
-    return create(uriInfo, securityContext, service, true);
+    Response response = create(uriInfo, securityContext, service, true);
+    decryptOrNullify(securityContext, (MessagingService) response.getEntity());
+    return response;
   }
 
   @PUT
@@ -360,22 +358,13 @@ public class MessagingServiceResource extends EntityResource<MessagingService, M
         .withServiceType(create.getServiceType());
   }
 
-  private ResultList<MessagingService> decryptOrNullify(
+  @Override
-      SecurityContext securityContext, ResultList<MessagingService> messagingServices) {
+  protected MessagingService nullifyConnection(MessagingService service) {
-    Optional.ofNullable(messagingServices.getData())
+    return service.withConnection(null);
-        .orElse(Collections.emptyList())
-        .forEach(messagingService -> decryptOrNullify(securityContext, messagingService));
-    return messagingServices;
   }
 
-  private MessagingService decryptOrNullify(SecurityContext securityContext, MessagingService messagingService) {
+  @Override
-    try {
+  protected String extractServiceType(MessagingService service) {
-      authorizer.authorizeAdmin(securityContext, true);
+    return service.getServiceType().value();
-    } catch (AuthorizationException e) {
-      return messagingService.withConnection(null);
-    }
-    secretsManager.encryptOrDecryptServiceConnection(
-        messagingService.getConnection(), messagingService.getServiceType().value(), messagingService.getName(), false);
-    return messagingService;
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/mlmodel/MlModelServiceResource.java

@@ -20,9 +20,7 @@ import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -49,12 +47,12 @@ import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.ListFilter;
 import org.openmetadata.catalog.jdbi3.MlModelServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
-import org.openmetadata.catalog.resources.EntityResource;
+import org.openmetadata.catalog.resources.services.ServiceEntityResource;
 import org.openmetadata.catalog.secrets.SecretsManager;
-import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.type.EntityHistory;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.type.MlModelConnection;
 import org.openmetadata.catalog.util.JsonUtils;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.ResultList;
@@ -64,13 +62,12 @@ import org.openmetadata.catalog.util.ResultList;
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
 @Collection(name = "mlmodelServices")
-public class MlModelServiceResource extends EntityResource<MlModelService, MlModelServiceRepository> {
+public class MlModelServiceResource
+    extends ServiceEntityResource<MlModelService, MlModelServiceRepository, MlModelConnection> {
   public static final String COLLECTION_PATH = "v1/services/mlmodelServices/";
 
   public static final String FIELDS = "pipelines,owner";
 
-  private final SecretsManager secretsManager;
-
   @Override
   public MlModelService addHref(UriInfo uriInfo, MlModelService service) {
     service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
@@ -80,8 +77,7 @@ public class MlModelServiceResource extends EntityResource<MlModelService, MlMod
   }
 
   public MlModelServiceResource(CollectionDAO dao, Authorizer authorizer, SecretsManager secretsManager) {
-    super(MlModelService.class, new MlModelServiceRepository(dao, secretsManager), authorizer);
+    super(MlModelService.class, new MlModelServiceRepository(dao, secretsManager), authorizer, secretsManager);
-    this.secretsManager = secretsManager;
   }
 
   public static class MlModelServiceList extends ResultList<MlModelService> {
@@ -295,7 +291,9 @@ public class MlModelServiceResource extends EntityResource<MlModelService, MlMod
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateMlModelService create)
       throws IOException {
     MlModelService service = getService(create, securityContext.getUserPrincipal().getName());
-    return create(uriInfo, securityContext, service, true);
+    Response response = create(uriInfo, securityContext, service, true);
+    decryptOrNullify(securityContext, (MlModelService) response.getEntity());
+    return response;
   }
 
   @PUT
@@ -316,7 +314,9 @@ public class MlModelServiceResource extends EntityResource<MlModelService, MlMod
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateMlModelService update)
       throws IOException {
     MlModelService service = getService(update, securityContext.getUserPrincipal().getName());
-    return createOrUpdate(uriInfo, securityContext, service, true);
+    Response response = createOrUpdate(uriInfo, securityContext, service, true);
+    decryptOrNullify(securityContext, (MlModelService) response.getEntity());
+    return response;
   }
 
   @DELETE
@@ -354,22 +354,13 @@ public class MlModelServiceResource extends EntityResource<MlModelService, MlMod
         .withConnection(create.getConnection());
   }
 
-  private ResultList<MlModelService> decryptOrNullify(
+  @Override
-      SecurityContext securityContext, ResultList<MlModelService> mlModelServices) {
+  protected MlModelService nullifyConnection(MlModelService service) {
-    Optional.ofNullable(mlModelServices.getData())
+    return service.withConnection(null);
-        .orElse(Collections.emptyList())
-        .forEach(mlModelService -> decryptOrNullify(securityContext, mlModelService));
-    return mlModelServices;
   }
 
-  private MlModelService decryptOrNullify(SecurityContext securityContext, MlModelService mlModelService) {
+  @Override
-    try {
+  protected String extractServiceType(MlModelService service) {
-      authorizer.authorizeAdmin(securityContext, true);
+    return service.getServiceType().value();
-    } catch (AuthorizationException e) {
-      return mlModelService.withConnection(null);
-    }
-    secretsManager.encryptOrDecryptServiceConnection(
-        mlModelService.getConnection(), mlModelService.getServiceType().value(), mlModelService.getName(), false);
-    return mlModelService;
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/services/pipeline/PipelineServiceResource.java

@@ -20,9 +20,7 @@ import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -49,12 +47,12 @@ import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.ListFilter;
 import org.openmetadata.catalog.jdbi3.PipelineServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
-import org.openmetadata.catalog.resources.EntityResource;
+import org.openmetadata.catalog.resources.services.ServiceEntityResource;
 import org.openmetadata.catalog.secrets.SecretsManager;
-import org.openmetadata.catalog.security.AuthorizationException;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.type.EntityHistory;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.type.PipelineConnection;
 import org.openmetadata.catalog.util.JsonUtils;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.ResultList;
@@ -64,12 +62,11 @@ import org.openmetadata.catalog.util.ResultList;
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
 @Collection(name = "pipelineServices")
-public class PipelineServiceResource extends EntityResource<PipelineService, PipelineServiceRepository> {
+public class PipelineServiceResource
+    extends ServiceEntityResource<PipelineService, PipelineServiceRepository, PipelineConnection> {
   public static final String COLLECTION_PATH = "v1/services/pipelineServices/";
   static final String FIELDS = "pipelines,owner";
 
-  private final SecretsManager secretsManager;
-
   @Override
   public PipelineService addHref(UriInfo uriInfo, PipelineService service) {
     service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
@@ -79,8 +76,7 @@ public class PipelineServiceResource extends EntityResource<PipelineService, Pip
   }
 
   public PipelineServiceResource(CollectionDAO dao, Authorizer authorizer, SecretsManager secretsManager) {
-    super(PipelineService.class, new PipelineServiceRepository(dao, secretsManager), authorizer);
+    super(PipelineService.class, new PipelineServiceRepository(dao, secretsManager), authorizer, secretsManager);
-    this.secretsManager = secretsManager;
   }
 
   public static class PipelineServiceList extends ResultList<PipelineService> {
@@ -357,22 +353,13 @@ public class PipelineServiceResource extends EntityResource<PipelineService, Pip
         .withConnection(create.getConnection());
   }
 
-  private ResultList<PipelineService> decryptOrNullify(
+  @Override
-      SecurityContext securityContext, ResultList<PipelineService> pipelineServices) {
+  protected PipelineService nullifyConnection(PipelineService service) {
-    Optional.ofNullable(pipelineServices.getData())
+    return service.withConnection(null);
-        .orElse(Collections.emptyList())
-        .forEach(pipelineService -> decryptOrNullify(securityContext, pipelineService));
-    return pipelineServices;
   }
 
-  private PipelineService decryptOrNullify(SecurityContext securityContext, PipelineService pipelineService) {
+  @Override
-    try {
+  protected String extractServiceType(PipelineService service) {
-      authorizer.authorizeAdmin(securityContext, true);
+    return service.getServiceType().value();
-    } catch (AuthorizationException e) {
-      return pipelineService.withConnection(null);
-    }
-    secretsManager.encryptOrDecryptServiceConnection(
-        pipelineService.getConnection(), pipelineService.getServiceType().value(), pipelineService.getName(), false);
-    return pipelineService;
   }
 }

catalog-rest-service/src/main/resources/json/schema/entity/services/dashboardService.json

@@ -145,6 +145,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

catalog-rest-service/src/main/resources/json/schema/entity/services/databaseService.json

@@ -284,6 +284,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

catalog-rest-service/src/main/resources/json/schema/entity/services/messagingService.json

@@ -115,6 +115,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

catalog-rest-service/src/main/resources/json/schema/entity/services/metadataService.json

@@ -104,6 +104,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

catalog-rest-service/src/main/resources/json/schema/entity/services/mlmodelService.json

@@ -107,6 +107,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

catalog-rest-service/src/main/resources/json/schema/entity/services/pipelineService.json

@@ -113,6 +113,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/DashboardServiceResourceUnitTest.java

@@ -0,0 +1,74 @@
+/*
+ *  Copyright 2022 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.services;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.openmetadata.catalog.Entity.FIELD_OWNER;
+
+import java.io.IOException;
+import java.util.UUID;
+import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.api.services.CreateDashboardService;
+import org.openmetadata.catalog.entity.services.DashboardService;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.DashboardServiceRepository;
+import org.openmetadata.catalog.resources.services.dashboard.DashboardServiceResource;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.DashboardConnection;
+import org.openmetadata.catalog.type.Include;
+
+public class DashboardServiceResourceUnitTest
+    extends ServiceResourceTest<
+        DashboardServiceResource, DashboardService, DashboardServiceRepository, DashboardConnection> {
+
+  @Override
+  protected DashboardServiceResource newServiceResource(
+      CollectionDAO collectionDAO, Authorizer authorizer, SecretsManager secretsManager) {
+    return new DashboardServiceResource(collectionDAO, authorizer, secretsManager);
+  }
+
+  @Override
+  protected void mockServiceResourceSpecific() throws IOException {
+    service = mock(DashboardService.class);
+    CollectionDAO.DashboardServiceDAO entityDAO = mock(CollectionDAO.DashboardServiceDAO.class);
+    when(collectionDAO.dashboardServiceDAO()).thenReturn(entityDAO);
+    lenient().when(service.getServiceType()).thenReturn(CreateDashboardService.DashboardServiceType.Tableau);
+    lenient().when(service.getConnection()).thenReturn(mock(DashboardConnection.class));
+    lenient().when(service.withConnection(isNull())).thenReturn(service);
+    when(entityDAO.findEntityById(any(), any())).thenReturn(service);
+    when(entityDAO.getEntityClass()).thenReturn(DashboardService.class);
+  }
+
+  @Override
+  protected String serviceType() {
+    return CreateDashboardService.DashboardServiceType.Tableau.value();
+  }
+
+  @Override
+  protected void verifyServiceWithConnectionCall(boolean shouldBeNull, DashboardService service) {
+    verify(service, times(shouldBeNull ? 1 : 0)).withConnection(isNull());
+  }
+
+  @Override
+  protected DashboardService callGetFromResource(DashboardServiceResource resource) throws IOException {
+    return resource.get(mock(UriInfo.class), securityContext, UUID.randomUUID().toString(), FIELD_OWNER, Include.ALL);
+  }
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/DatabaseServiceResourceUnitTest.java

@@ -0,0 +1,74 @@
+/*
+ *  Copyright 2022 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.services;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.openmetadata.catalog.Entity.FIELD_OWNER;
+
+import java.io.IOException;
+import java.util.UUID;
+import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.api.services.CreateDatabaseService;
+import org.openmetadata.catalog.api.services.DatabaseConnection;
+import org.openmetadata.catalog.entity.services.DatabaseService;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.DatabaseServiceRepository;
+import org.openmetadata.catalog.resources.services.database.DatabaseServiceResource;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.Include;
+
+public class DatabaseServiceResourceUnitTest
+    extends ServiceResourceTest<
+        DatabaseServiceResource, DatabaseService, DatabaseServiceRepository, DatabaseConnection> {
+
+  @Override
+  protected DatabaseServiceResource newServiceResource(
+      CollectionDAO collectionDAO, Authorizer authorizer, SecretsManager secretsManager) {
+    return new DatabaseServiceResource(collectionDAO, authorizer, secretsManager);
+  }
+
+  @Override
+  protected void mockServiceResourceSpecific() throws IOException {
+    service = mock(DatabaseService.class);
+    CollectionDAO.DatabaseServiceDAO entityDAO = mock(CollectionDAO.DatabaseServiceDAO.class);
+    when(collectionDAO.dbServiceDAO()).thenReturn(entityDAO);
+    lenient().when(service.getServiceType()).thenReturn(CreateDatabaseService.DatabaseServiceType.Mysql);
+    lenient().when(service.getConnection()).thenReturn(mock(DatabaseConnection.class));
+    lenient().when(service.withConnection(isNull())).thenReturn(service);
+    when(entityDAO.findEntityById(any(), any())).thenReturn(service);
+    when(entityDAO.getEntityClass()).thenReturn(DatabaseService.class);
+  }
+
+  @Override
+  protected String serviceType() {
+    return CreateDatabaseService.DatabaseServiceType.Mysql.value();
+  }
+
+  @Override
+  protected void verifyServiceWithConnectionCall(boolean shouldBeNull, DatabaseService service) {
+    verify(service, times(shouldBeNull ? 1 : 0)).withConnection(isNull());
+  }
+
+  @Override
+  protected DatabaseService callGetFromResource(DatabaseServiceResource resource) throws IOException {
+    return resource.get(mock(UriInfo.class), securityContext, UUID.randomUUID().toString(), FIELD_OWNER, Include.ALL);
+  }
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/MessagingServiceResourceUnitTest.java

@@ -0,0 +1,74 @@
+/*
+ *  Copyright 2022 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.services;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.openmetadata.catalog.Entity.FIELD_OWNER;
+
+import java.io.IOException;
+import java.util.UUID;
+import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.api.services.CreateMessagingService;
+import org.openmetadata.catalog.entity.services.MessagingService;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.MessagingServiceRepository;
+import org.openmetadata.catalog.resources.services.messaging.MessagingServiceResource;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.type.MessagingConnection;
+
+public class MessagingServiceResourceUnitTest
+    extends ServiceResourceTest<
+        MessagingServiceResource, MessagingService, MessagingServiceRepository, MessagingConnection> {
+
+  @Override
+  protected MessagingServiceResource newServiceResource(
+      CollectionDAO collectionDAO, Authorizer authorizer, SecretsManager secretsManager) {
+    return new MessagingServiceResource(collectionDAO, authorizer, secretsManager);
+  }
+
+  @Override
+  protected void mockServiceResourceSpecific() throws IOException {
+    service = mock(MessagingService.class);
+    CollectionDAO.MessagingServiceDAO entityDAO = mock(CollectionDAO.MessagingServiceDAO.class);
+    when(collectionDAO.messagingServiceDAO()).thenReturn(entityDAO);
+    lenient().when(service.getServiceType()).thenReturn(CreateMessagingService.MessagingServiceType.Kafka);
+    lenient().when(service.getConnection()).thenReturn(mock(MessagingConnection.class));
+    lenient().when(service.withConnection(isNull())).thenReturn(service);
+    when(entityDAO.findEntityById(any(), any())).thenReturn(service);
+    when(entityDAO.getEntityClass()).thenReturn(MessagingService.class);
+  }
+
+  @Override
+  protected String serviceType() {
+    return CreateMessagingService.MessagingServiceType.Kafka.value();
+  }
+
+  @Override
+  protected void verifyServiceWithConnectionCall(boolean shouldBeNull, MessagingService service) {
+    verify(service, times(shouldBeNull ? 1 : 0)).withConnection(isNull());
+  }
+
+  @Override
+  protected MessagingService callGetFromResource(MessagingServiceResource resource) throws IOException {
+    return resource.get(mock(UriInfo.class), securityContext, UUID.randomUUID().toString(), FIELD_OWNER, Include.ALL);
+  }
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/MlModelServiceResourceUnitTest.java

@@ -0,0 +1,76 @@
+/*
+ *  Copyright 2022 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.services;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.openmetadata.catalog.Entity.FIELD_OWNER;
+
+import java.io.IOException;
+import java.util.UUID;
+import javax.ws.rs.core.UriInfo;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.openmetadata.catalog.api.services.CreateMlModelService;
+import org.openmetadata.catalog.entity.services.MlModelService;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.MlModelServiceRepository;
+import org.openmetadata.catalog.resources.services.mlmodel.MlModelServiceResource;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.type.MlModelConnection;
+
+@ExtendWith(MockitoExtension.class)
+public class MlModelServiceResourceUnitTest
+    extends ServiceResourceTest<MlModelServiceResource, MlModelService, MlModelServiceRepository, MlModelConnection> {
+
+  @Override
+  protected MlModelServiceResource newServiceResource(
+      CollectionDAO collectionDAO, Authorizer authorizer, SecretsManager secretsManager) {
+    return new MlModelServiceResource(collectionDAO, authorizer, secretsManager);
+  }
+
+  @Override
+  protected void mockServiceResourceSpecific() throws IOException {
+    service = mock(MlModelService.class);
+    CollectionDAO.MlModelServiceDAO entityDAO = mock(CollectionDAO.MlModelServiceDAO.class);
+    when(collectionDAO.mlModelServiceDAO()).thenReturn(entityDAO);
+    lenient().when(service.getServiceType()).thenReturn(CreateMlModelService.MlModelServiceType.Mlflow);
+    lenient().when(service.getConnection()).thenReturn(mock(MlModelConnection.class));
+    lenient().when(service.withConnection(isNull())).thenReturn(service);
+    when(entityDAO.findEntityById(any(), any())).thenReturn(service);
+    when(entityDAO.getEntityClass()).thenReturn(MlModelService.class);
+  }
+
+  @Override
+  protected String serviceType() {
+    return CreateMlModelService.MlModelServiceType.Mlflow.value();
+  }
+
+  @Override
+  protected void verifyServiceWithConnectionCall(boolean shouldBeNull, MlModelService service) {
+    verify(service, times(shouldBeNull ? 1 : 0)).withConnection(isNull());
+  }
+
+  @Override
+  protected MlModelService callGetFromResource(MlModelServiceResource resource) throws IOException {
+    return resource.get(mock(UriInfo.class), securityContext, UUID.randomUUID().toString(), FIELD_OWNER, Include.ALL);
+  }
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/PipelineServiceResourceUnitTest.java

@@ -0,0 +1,77 @@
+/*
+ *  Copyright 2022 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.services;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.openmetadata.catalog.Entity.FIELD_OWNER;
+
+import java.io.IOException;
+import java.util.UUID;
+import javax.ws.rs.core.UriInfo;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.openmetadata.catalog.api.services.CreatePipelineService;
+import org.openmetadata.catalog.entity.services.PipelineService;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.PipelineServiceRepository;
+import org.openmetadata.catalog.resources.services.pipeline.PipelineServiceResource;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.type.PipelineConnection;
+
+@ExtendWith(MockitoExtension.class)
+public class PipelineServiceResourceUnitTest
+    extends ServiceResourceTest<
+        PipelineServiceResource, PipelineService, PipelineServiceRepository, PipelineConnection> {
+
+  @Override
+  protected PipelineServiceResource newServiceResource(
+      CollectionDAO collectionDAO, Authorizer authorizer, SecretsManager secretsManager) {
+    return new PipelineServiceResource(collectionDAO, authorizer, secretsManager);
+  }
+
+  @Override
+  protected void mockServiceResourceSpecific() throws IOException {
+    service = mock(PipelineService.class);
+    CollectionDAO.PipelineServiceDAO entityDAO = mock(CollectionDAO.PipelineServiceDAO.class);
+    when(collectionDAO.pipelineServiceDAO()).thenReturn(entityDAO);
+    lenient().when(service.getServiceType()).thenReturn(CreatePipelineService.PipelineServiceType.Airflow);
+    lenient().when(service.getConnection()).thenReturn(mock(PipelineConnection.class));
+    lenient().when(service.withConnection(isNull())).thenReturn(service);
+    when(entityDAO.findEntityById(any(), any())).thenReturn(service);
+    when(entityDAO.getEntityClass()).thenReturn(PipelineService.class);
+  }
+
+  @Override
+  protected String serviceType() {
+    return CreatePipelineService.PipelineServiceType.Airflow.value();
+  }
+
+  @Override
+  protected void verifyServiceWithConnectionCall(boolean shouldBeNull, PipelineService service) {
+    verify(service, times(shouldBeNull ? 1 : 0)).withConnection(isNull());
+  }
+
+  @Override
+  protected PipelineService callGetFromResource(PipelineServiceResource resource) throws IOException {
+    return resource.get(mock(UriInfo.class), securityContext, UUID.randomUUID().toString(), FIELD_OWNER, Include.ALL);
+  }
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/ServiceResourceTest.java

@@ -0,0 +1,113 @@
+package org.openmetadata.catalog.resources.services;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.notNull;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.UUID;
+import javax.ws.rs.core.SecurityContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.openmetadata.catalog.ServiceConnectionEntityInterface;
+import org.openmetadata.catalog.ServiceEntityInterface;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.ServiceRepository;
+import org.openmetadata.catalog.secrets.SecretsManager;
+import org.openmetadata.catalog.security.AuthorizationException;
+import org.openmetadata.catalog.security.Authorizer;
+
+@ExtendWith(MockitoExtension.class)
+public abstract class ServiceResourceTest<
+    T extends ServiceEntityResource<R, S, U>,
+    R extends ServiceEntityInterface,
+    S extends ServiceRepository<R, U>,
+    U extends ServiceConnectionEntityInterface> {
+
+  T serviceResource;
+
+  @Mock protected CollectionDAO collectionDAO;
+
+  @Mock protected Authorizer authorizer;
+
+  @Mock protected SecretsManager secretsManager;
+
+  protected R service;
+
+  @Mock protected SecurityContext securityContext;
+
+  @BeforeEach
+  void beforeEach() throws IOException {
+    mockServiceResourceSpecific();
+    when(collectionDAO.relationshipDAO()).thenReturn(mock(CollectionDAO.EntityRelationshipDAO.class));
+    when(service.getId()).thenReturn(UUID.randomUUID());
+    when(service.withHref(any())).thenReturn(service);
+    lenient()
+        .when(
+            secretsManager.encryptOrDecryptServiceConnectionConfig(
+                any(), anyString(), anyString(), anyString(), anyBoolean()))
+        .thenReturn(service);
+    serviceResource = newServiceResource(collectionDAO, authorizer, secretsManager);
+  }
+
+  protected abstract void mockServiceResourceSpecific() throws IOException;
+
+  protected abstract T newServiceResource(
+      CollectionDAO collectionDAO, Authorizer authorizer, SecretsManager secretsManager);
+
+  @ParameterizedTest
+  @CsvSource({
+    "true, true, true, false",
+    "true, true, false, false",
+    "true, false, true, false",
+    "true, false, false, true",
+    "false, true, true, false",
+    "false, true, false, false",
+    "false, false, true, true",
+    "false, false, false, true"
+  })
+  void testGetCallDecryptOrNullify(boolean isLocalSecretManager, boolean isAdmin, boolean isBot, boolean shouldBeNull)
+      throws IOException {
+    lenient().when(secretsManager.isLocal()).thenReturn(isLocalSecretManager);
+
+    if (isLocalSecretManager && !isAdmin && !isBot) {
+      lenient()
+          .doThrow(new AuthorizationException(""))
+          .when(authorizer)
+          .authorizeAdmin(any(SecurityContext.class), eq(true));
+    } else if (!isLocalSecretManager && !isAdmin) {
+      lenient()
+          .doThrow(new AuthorizationException(""))
+          .when(authorizer)
+          .authorizeAdmin(any(SecurityContext.class), eq(false));
+    }
+
+    R actual = callGetFromResource(serviceResource);
+
+    verify(secretsManager, times(1)).isLocal();
+    verify(secretsManager, times(shouldBeNull ? 0 : 1))
+        .encryptOrDecryptServiceConnection(notNull(), eq(serviceType()), any(), eq(false));
+
+    verifyServiceWithConnectionCall(shouldBeNull, service);
+
+    assertEquals(service, actual);
+  }
+
+  protected abstract String serviceType();
+
+  protected abstract void verifyServiceWithConnectionCall(boolean shouldBeNull, R service);
+
+  protected abstract R callGetFromResource(T resource) throws IOException;
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/services/ingestionpipelines/IngestionPipelineResourceUnitTest.java

@@ -0,0 +1,95 @@
+/*
+ *  Copyright 2022 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.services.ingestionpipelines;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import javax.ws.rs.core.SecurityContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedConstruction.Context;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.openmetadata.catalog.CatalogApplicationConfig;
+import org.openmetadata.catalog.airflow.AirflowRESTClient;
+import org.openmetadata.catalog.entity.services.ingestionPipelines.IngestionPipeline;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.security.Authorizer;
+import org.openmetadata.catalog.type.EntityReference;
+import org.openmetadata.catalog.util.PipelineServiceClient;
+
+@ExtendWith(MockitoExtension.class)
+public class IngestionPipelineResourceUnitTest {
+
+  private static final String DAG_ID = UUID.randomUUID().toString();
+
+  private IngestionPipelineResource ingestionPipelineResource;
+
+  @Mock SecurityContext securityContext;
+
+  @Mock Authorizer authorizer;
+
+  @Mock CollectionDAO collectionDAO;
+
+  @Mock CatalogApplicationConfig catalogApplicationConfig;
+
+  @Mock IngestionPipeline ingestionPipeline;
+
+  @BeforeEach
+  void setUp() throws IOException {
+    CollectionDAO.IngestionPipelineDAO entityDAO = mock(CollectionDAO.IngestionPipelineDAO.class);
+    CollectionDAO.EntityRelationshipDAO relationshipDAO = mock(CollectionDAO.EntityRelationshipDAO.class);
+    CollectionDAO.EntityRelationshipRecord entityRelationshipRecord =
+        mock(CollectionDAO.EntityRelationshipRecord.class);
+    when(entityRelationshipRecord.getId()).thenReturn(UUID.randomUUID());
+    when(entityRelationshipRecord.getType()).thenReturn("ingestionPipeline");
+    when(relationshipDAO.findFrom(any(), any(), anyInt())).thenReturn(List.of(entityRelationshipRecord));
+    when(collectionDAO.ingestionPipelineDAO()).thenReturn(entityDAO);
+    when(collectionDAO.relationshipDAO()).thenReturn(relationshipDAO);
+    when(entityDAO.findEntityById(any(), any())).thenReturn(ingestionPipeline);
+    when(entityDAO.findEntityReferenceById(any(), any())).thenReturn(mock(EntityReference.class));
+    when(entityDAO.getEntityClass()).thenReturn(IngestionPipeline.class);
+    when(ingestionPipeline.getId()).thenReturn(UUID.fromString(DAG_ID));
+    ingestionPipelineResource = new IngestionPipelineResource(collectionDAO, authorizer);
+  }
+
+  @Test
+  public void testLastIngestionLogsAreRetrieved() throws IOException {
+    Map<String, String> expectedMap = Map.of("task", "log");
+    try (MockedConstruction<AirflowRESTClient> mocked =
+        mockConstruction(AirflowRESTClient.class, this::preparePipelineServiceClient)) {
+      ingestionPipelineResource.initialize(catalogApplicationConfig);
+      assertEquals(
+          expectedMap, ingestionPipelineResource.getLastIngestionLogs(null, securityContext, DAG_ID).getEntity());
+      PipelineServiceClient client = mocked.constructed().get(0);
+      verify(client).getLastIngestionLogs(ingestionPipeline);
+    }
+  }
+
+  private void preparePipelineServiceClient(AirflowRESTClient mockPipelineServiceClient, Context context) {
+    when(mockPipelineServiceClient.getLastIngestionLogs(any())).thenReturn(Map.of("task", "log"));
+  }
+}

ingestion/src/metadata/utils/secrets_manager.py

@@ -8,112 +8,112 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
+import inspect
 import json
 from abc import abstractmethod
 from pydoc import locate
-from typing import Optional, Union
+from typing import NewType, Optional, Union
 
 import boto3
 from botocore.exceptions import ClientError
+from pydantic.main import ModelMetaclass
 
 from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
     SecretsManagerProvider,
 )
-from metadata.generated.schema.entity.services.dashboardService import DashboardService
+from metadata.generated.schema.entity.services.connections.serviceConnection import (
-from metadata.generated.schema.entity.services.databaseService import DatabaseService
+    ServiceConnection,
-from metadata.generated.schema.entity.services.messagingService import MessagingService
+)
-from metadata.generated.schema.entity.services.mlmodelService import MlModelService
-from metadata.generated.schema.entity.services.pipelineService import PipelineService
 from metadata.generated.schema.security.credentials.awsCredentials import AWSCredentials
 from metadata.utils.logger import ingestion_logger
 from metadata.utils.singleton import Singleton
 
 logger = ingestion_logger()
+
 SECRET_MANAGER_AIRFLOW_CONF = "openmetadata_secrets_manager"
 
+ServiceConnectionType = NewType(
+    "ServiceConnectionType", ServiceConnection.__fields__["__root__"].type_
+)
+
 
 class SecretsManager(metaclass=Singleton):
     @abstractmethod
     def add_service_config_connection(
         self,
-        service: Union[
+        service: ServiceConnectionType,
-            DashboardService,
-            DatabaseService,
-            MessagingService,
-            PipelineService,
-            MlModelService,
-        ],
         service_type: str,
-    ) -> Union[
+    ) -> ServiceConnectionType:
-        DashboardService,
-        DatabaseService,
-        MessagingService,
-        PipelineService,
-        MlModelService,
-    ]:
         pass
 
     @staticmethod
+    def to_service_simple(service_type: str) -> str:
+        return service_type.replace("Service", "").lower()
+
     def build_secret_id(
-        service_type: str, service_connection_type: str, service_name: str
+        self, service_type: str, service_connection_type: str, service_name: str
     ) -> str:
-        return f"openmetadata-{service_type.lower()}-{service_connection_type.lower()}-{service_name.lower()}"
+        return f"openmetadata-{self.to_service_simple(service_type).lower()}-{service_connection_type.lower()}-{service_name.lower()}"
+
+    def get_service_connection_class(self, service_type) -> ModelMetaclass:
+        service_conn_name = next(
+            (
+                clazz[1]
+                for clazz in inspect.getmembers(
+                    locate(f"metadata.generated.schema.entity.services.{service_type}"),
+                    inspect.isclass,
+                )
+                if clazz[0].lower()
+                == f"{self.to_service_simple(service_type)}connection"
+            )
+        ).__name__
+        service_conn_class = locate(
+            f"metadata.generated.schema.entity.services.{service_type}.{service_conn_name}"
+        )
+        return service_conn_class
+
+    def get_connection_class(
+        self, service_type: str, service_connection_type
+    ) -> ModelMetaclass:
+        connection_py_file = (
+            service_connection_type[0].lower() + service_connection_type[1:]
+        )
+        conn_class = locate(
+            f"metadata.generated.schema.entity.services.connections.{self.to_service_simple(service_type)}.{connection_py_file}Connection.{service_connection_type}Connection"
+        )
+        return conn_class
 
 
 class LocalSecretsManager(SecretsManager):
     def add_service_config_connection(
         self,
-        service: Union[
+        service: ServiceConnectionType,
-            DashboardService,
-            DatabaseService,
-            MessagingService,
-            PipelineService,
-            MlModelService,
-        ],
         service_type: str,
-    ) -> Union[
+    ) -> ServiceConnectionType:
-        DashboardService,
-        DatabaseService,
-        MessagingService,
-        PipelineService,
-        MlModelService,
-    ]:
         return service
 
 
 class AWSSecretsManager(SecretsManager):
     def add_service_config_connection(
         self,
-        service: Union[
+        service: ServiceConnectionType,
-            DashboardService,
-            DatabaseService,
-            MessagingService,
-            PipelineService,
-            MlModelService,
-        ],
         service_type: str,
-    ) -> Union[
+    ) -> ServiceConnectionType:
-        DashboardService,
-        DatabaseService,
-        MessagingService,
-        PipelineService,
-        MlModelService,
-    ]:
-        service_simple_type = service_type.replace("Service", "").lower()
         service_connection_type = service.serviceType.value
         service_name = service.name.__root__
         secret_id = self.build_secret_id(
-            service_simple_type, service_connection_type, service_name
+            service_type, service_connection_type, service_name
         )
-        connection_py_file = (
+        connection_class = self.get_connection_class(
-            service_connection_type[0].lower() + service_connection_type[1:]
+            service_type, service_connection_type
         )
-        conn_class = locate(
+        service_conn_class = self.get_service_connection_class(service_type)
-            f"metadata.generated.schema.entity.services.connections.{service_simple_type}.{connection_py_file}Connection.{service_connection_type}Connection"
+        service.connection = service_conn_class(
-        )
+            config=connection_class.parse_obj(
-        service.connection.config = conn_class.parse_obj(
                 json.loads(self._get_string_value(secret_id))
             )
+        )
+
         return service
 
     def __init__(self, credentials: AWSCredentials):

ingestion/tests/unit/metadata/utils/test_secrets_manager.py

@@ -82,6 +82,8 @@ class TestSecretsManager(TestCase):
         expected_service = deepcopy(self.service)
         expected_service.connection.config = self.database_connection
 
+        self.service.connection = None
+
         aws_manager.add_service_config_connection(self.service, self.service_type)
 
         self.assertEqual(expected_service, self.service)

openmetadata-core/src/main/resources/json/schema/entity/services/dashboardService.json

@@ -140,6 +140,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

openmetadata-core/src/main/resources/json/schema/entity/services/databaseService.json

@@ -279,6 +279,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

openmetadata-core/src/main/resources/json/schema/entity/services/messagingService.json

@@ -110,6 +110,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

openmetadata-core/src/main/resources/json/schema/entity/services/metadataService.json

@@ -104,6 +104,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

openmetadata-core/src/main/resources/json/schema/entity/services/mlmodelService.json

@@ -102,6 +102,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }

openmetadata-core/src/main/resources/json/schema/entity/services/pipelineService.json

@@ -114,6 +114,6 @@
       "default": false
     }
   },
-  "required": ["id", "name", "serviceType", "connection"],
+  "required": ["id", "name", "serviceType"],
   "additionalProperties": false
 }