Fix #2434: All services must have owner (#2520)

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/DashboardServiceRepository.java

@@ -13,11 +13,13 @@
 
 package org.openmetadata.catalog.jdbi3;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import java.io.IOException;
 import java.net.URI;
+import java.text.ParseException;
 import java.util.UUID;
-import javax.ws.rs.core.UriInfo;
 import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.entity.services.DashboardService;
 import org.openmetadata.catalog.resources.services.dashboard.DashboardServiceResource;
@@ -27,9 +29,10 @@ import org.openmetadata.catalog.type.Schedule;
 import org.openmetadata.catalog.util.EntityInterface;
 import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.EntityUtil.Fields;
-import org.openmetadata.catalog.util.JsonUtils;
 
 public class DashboardServiceRepository extends EntityRepository<DashboardService> {
+  private static final Fields UPDATE_FIELDS = new Fields(DashboardServiceResource.FIELD_LIST, "owner");
+
   public DashboardServiceRepository(CollectionDAO dao) {
     super(
         DashboardServiceResource.COLLECTION_PATH,
@@ -38,36 +41,15 @@ public class DashboardServiceRepository extends EntityRepository<DashboardServic
         dao.dashboardServiceDAO(),
         dao,
         Fields.EMPTY_FIELDS,
-        Fields.EMPTY_FIELDS,
+        UPDATE_FIELDS,
-        false,
         false,
+        true,
         false);
   }
 
-  public DashboardService update(
-      UriInfo uriInfo,
-      UUID id,
-      String description,
-      URI dashboardUrl,
-      String username,
-      String password,
-      Schedule ingestionSchedule)
-      throws IOException {
-    EntityUtil.validateIngestionSchedule(ingestionSchedule);
-    DashboardService dashboardService = daoCollection.dashboardServiceDAO().findEntityById(id);
-    // Update fields
-    dashboardService
-        .withDescription(description)
-        .withDashboardUrl(dashboardUrl)
-        .withUsername(username)
-        .withPassword(password)
-        .withIngestionSchedule(ingestionSchedule);
-    daoCollection.dashboardServiceDAO().update(id, JsonUtils.pojoToJson(dashboardService));
-    return withHref(uriInfo, dashboardService);
-  }
-
   @Override
-  public DashboardService setFields(DashboardService entity, Fields fields) {
+  public DashboardService setFields(DashboardService entity, Fields fields) throws IOException, ParseException {
+    entity.setOwner(fields.contains("owner") ? getOwner(entity) : null);
     return entity;
   }
 
@@ -80,21 +62,31 @@ public class DashboardServiceRepository extends EntityRepository<DashboardServic
   }
 
   @Override
-  public void prepare(DashboardService entity) {
+  public void prepare(DashboardService entity) throws IOException, ParseException {
+    // Check if owner is valid and set the relationship
+    entity.setOwner(helper(entity).validateOwnerOrNull());
     EntityUtil.validateIngestionSchedule(entity.getIngestionSchedule());
   }
 
   @Override
   public void storeEntity(DashboardService service, boolean update) throws IOException {
-    if (update) {
+    // Relationships and fields such as href are derived and not stored as part of json
-      daoCollection.dashboardServiceDAO().update(service.getId(), JsonUtils.pojoToJson(service));
+    EntityReference owner = service.getOwner();
-    } else {
+
-      daoCollection.dashboardServiceDAO().insert(service);
+    // Don't store owner, database, href and tags as JSON. Build it on the fly based on relationships
-    }
+    service.withOwner(null).withHref(null);
+
+    store(service.getId(), service, update);
+
+    // Restore the relationships
+    service.withOwner(owner);
   }
 
   @Override
-  public void storeRelationships(DashboardService entity) {}
+  public void storeRelationships(DashboardService entity) {
+    // Add owner relationship
+    setOwner(entity, entity.getOwner());
+  }
 
   @Override
   public EntityUpdater getUpdater(DashboardService original, DashboardService updated, Operation operation) {
@@ -128,6 +120,11 @@ public class DashboardServiceRepository extends EntityRepository<DashboardServic
       return entity.getDeleted();
     }
 
+    @Override
+    public EntityReference getOwner() {
+      return entity.getOwner();
+    }
+
     @Override
     public String getFullyQualifiedName() {
       return entity.getName();
@@ -200,6 +197,11 @@ public class DashboardServiceRepository extends EntityRepository<DashboardServic
       entity.setChangeDescription(changeDescription);
     }
 
+    @Override
+    public void setOwner(EntityReference owner) {
+      entity.setOwner(owner);
+    }
+
     @Override
     public void setDeleted(boolean flag) {
       entity.setDeleted(flag);

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/DatabaseServiceRepository.java

@@ -13,11 +13,13 @@
 
 package org.openmetadata.catalog.jdbi3;
 
+import static org.openmetadata.catalog.Entity.helper;
 import static org.openmetadata.catalog.util.EntityUtil.toBoolean;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import java.io.IOException;
 import java.net.URI;
+import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
@@ -31,7 +33,7 @@ import org.openmetadata.catalog.util.EntityInterface;
 import org.openmetadata.catalog.util.EntityUtil.Fields;
 
 public class DatabaseServiceRepository extends EntityRepository<DatabaseService> {
-  public static final String COLLECTION_PATH = "v1/services/databaseServices";
+  private static final Fields UPDATE_FIELDS = new Fields(DatabaseServiceResource.FIELD_LIST, "owner");
 
   public DatabaseServiceRepository(CollectionDAO dao) {
     super(
@@ -41,15 +43,16 @@ public class DatabaseServiceRepository extends EntityRepository<DatabaseService>
         dao.dbServiceDAO(),
         dao,
         Fields.EMPTY_FIELDS,
-        Fields.EMPTY_FIELDS,
+        UPDATE_FIELDS,
-        false,
         false,
+        true,
         false);
   }
 
   @Override
-  public DatabaseService setFields(DatabaseService entity, Fields fields) throws IOException {
+  public DatabaseService setFields(DatabaseService entity, Fields fields) throws IOException, ParseException {
     entity.setAirflowPipelines(fields.contains("airflowPipeline") ? getAirflowPipelines(entity) : null);
+    entity.setOwner(fields.contains("owner") ? getOwner(entity) : null);
     return entity;
   }
 
@@ -86,17 +89,29 @@ public class DatabaseServiceRepository extends EntityRepository<DatabaseService>
   }
 
   @Override
-  public void prepare(DatabaseService entity) {}
+  public void prepare(DatabaseService entity) throws IOException, ParseException {
+    // Check if owner is valid and set the relationship
+    entity.setOwner(helper(entity).validateOwnerOrNull());
+  }
 
   @Override
   public void storeEntity(DatabaseService service, boolean update) throws IOException {
-    service.withHref(null);
+    // Relationships and fields such as href are derived and not stored as part of json
+    EntityReference owner = service.getOwner();
+
+    // Don't store owner, database, href and tags as JSON. Build it on the fly based on relationships
+    service.withOwner(null).withHref(null);
+
     store(service.getId(), service, update);
+
+    // Restore the relationships
+    service.withOwner(owner);
   }
 
   @Override
   public void storeRelationships(DatabaseService entity) {
-    /* Nothing to do */
+    // Add owner relationship
+    setOwner(entity, entity.getOwner());
   }
 
   @Override
@@ -126,6 +141,11 @@ public class DatabaseServiceRepository extends EntityRepository<DatabaseService>
       return entity.getDisplayName();
     }
 
+    @Override
+    public EntityReference getOwner() {
+      return entity.getOwner();
+    }
+
     @Override
     public Boolean isDeleted() {
       return entity.getDeleted();
@@ -203,6 +223,11 @@ public class DatabaseServiceRepository extends EntityRepository<DatabaseService>
       entity.setChangeDescription(changeDescription);
     }
 
+    @Override
+    public void setOwner(EntityReference owner) {
+      entity.setOwner(owner);
+    }
+
     @Override
     public void setDeleted(boolean flag) {
       entity.setDeleted(flag);

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/MessagingServiceRepository.java

@@ -13,9 +13,12 @@
 
 package org.openmetadata.catalog.jdbi3;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import java.io.IOException;
 import java.net.URI;
+import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
@@ -29,6 +32,7 @@ import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.EntityUtil.Fields;
 
 public class MessagingServiceRepository extends EntityRepository<MessagingService> {
+  private static final Fields UPDATE_FIELDS = new Fields(MessagingServiceResource.FIELD_LIST, "owner");
 
   public MessagingServiceRepository(CollectionDAO dao) {
     super(
@@ -38,14 +42,15 @@ public class MessagingServiceRepository extends EntityRepository<MessagingServic
         dao.messagingServiceDAO(),
         dao,
         Fields.EMPTY_FIELDS,
-        Fields.EMPTY_FIELDS,
+        UPDATE_FIELDS,
-        false,
         false,
+        true,
         false);
   }
 
   @Override
-  public MessagingService setFields(MessagingService entity, Fields fields) {
+  public MessagingService setFields(MessagingService entity, Fields fields) throws IOException, ParseException {
+    entity.setOwner(fields.contains("owner") ? getOwner(entity) : null);
     return entity;
   }
 
@@ -60,19 +65,31 @@ public class MessagingServiceRepository extends EntityRepository<MessagingServic
   }
 
   @Override
-  public void prepare(MessagingService entity) {
+  public void prepare(MessagingService entity) throws IOException, ParseException {
+    // Check if owner is valid and set the relationship
+    entity.setOwner(helper(entity).validateOwnerOrNull());
+
     EntityUtil.validateIngestionSchedule(entity.getIngestionSchedule());
   }
 
   @Override
   public void storeEntity(MessagingService service, boolean update) throws IOException {
-    service.withHref(null);
+    // Relationships and fields such as href are derived and not stored as part of json
+    EntityReference owner = service.getOwner();
+
+    // Don't store owner, database, href and tags as JSON. Build it on the fly based on relationships
+    service.withOwner(null).withHref(null);
+
     store(service.getId(), service, update);
+
+    // Restore the relationships
+    service.withOwner(owner);
   }
 
   @Override
   public void storeRelationships(MessagingService entity) {
-    /* Nothing to do */
+    // Add owner relationship
+    setOwner(entity, entity.getOwner());
   }
 
   @Override
@@ -132,6 +149,11 @@ public class MessagingServiceRepository extends EntityRepository<MessagingServic
       return entity.getHref();
     }
 
+    @Override
+    public EntityReference getOwner() {
+      return entity.getOwner();
+    }
+
     @Override
     public ChangeDescription getChangeDescription() {
       return entity.getChangeDescription();
@@ -179,6 +201,11 @@ public class MessagingServiceRepository extends EntityRepository<MessagingServic
       entity.setChangeDescription(changeDescription);
     }
 
+    @Override
+    public void setOwner(EntityReference owner) {
+      entity.setOwner(owner);
+    }
+
     @Override
     public void setDeleted(boolean flag) {
       entity.setDeleted(flag);

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/PipelineServiceRepository.java

@@ -13,8 +13,11 @@
 
 package org.openmetadata.catalog.jdbi3;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import java.io.IOException;
 import java.net.URI;
+import java.text.ParseException;
 import java.util.UUID;
 import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.entity.services.PipelineService;
@@ -24,9 +27,10 @@ import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.util.EntityInterface;
 import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.EntityUtil.Fields;
-import org.openmetadata.catalog.util.JsonUtils;
 
 public class PipelineServiceRepository extends EntityRepository<PipelineService> {
+  private static final Fields UPDATE_FIELDS = new Fields(PipelineServiceResource.FIELD_LIST, "owner");
+
   public PipelineServiceRepository(CollectionDAO dao) {
     super(
         PipelineServiceResource.COLLECTION_PATH,
@@ -35,14 +39,15 @@ public class PipelineServiceRepository extends EntityRepository<PipelineService>
         dao.pipelineServiceDAO(),
         dao,
         Fields.EMPTY_FIELDS,
-        Fields.EMPTY_FIELDS,
+        UPDATE_FIELDS,
-        false,
         false,
+        true,
         false);
   }
 
   @Override
-  public PipelineService setFields(PipelineService entity, Fields fields) {
+  public PipelineService setFields(PipelineService entity, Fields fields) throws IOException, ParseException {
+    entity.setOwner(fields.contains("owner") ? getOwner(entity) : null);
     return entity;
   }
 
@@ -57,22 +62,30 @@ public class PipelineServiceRepository extends EntityRepository<PipelineService>
   }
 
   @Override
-  public void prepare(PipelineService entity) {
+  public void prepare(PipelineService entity) throws IOException, ParseException {
+    // Check if owner is valid and set the relationship
+    entity.setOwner(helper(entity).validateOwnerOrNull());
     EntityUtil.validateIngestionSchedule(entity.getIngestionSchedule());
   }
 
   @Override
   public void storeEntity(PipelineService service, boolean update) throws IOException {
-    if (update) {
+    // Relationships and fields such as href are derived and not stored as part of json
-      daoCollection.pipelineServiceDAO().update(service.getId(), JsonUtils.pojoToJson(service));
+    EntityReference owner = service.getOwner();
-    } else {
+
-      daoCollection.pipelineServiceDAO().insert(service);
+    // Don't store owner, database, href and tags as JSON. Build it on the fly based on relationships
-    }
+    service.withOwner(null).withHref(null);
+
+    store(service.getId(), service, update);
+
+    // Restore the relationships
+    service.withOwner(owner);
   }
 
   @Override
   public void storeRelationships(PipelineService entity) {
-    /* Nothing to do */
+    // Add owner relationship
+    setOwner(entity, entity.getOwner());
   }
 
   @Override
@@ -132,6 +145,11 @@ public class PipelineServiceRepository extends EntityRepository<PipelineService>
       return entity.getHref();
     }
 
+    @Override
+    public EntityReference getOwner() {
+      return entity.getOwner();
+    }
+
     @Override
     public ChangeDescription getChangeDescription() {
       return entity.getChangeDescription();
@@ -179,6 +197,11 @@ public class PipelineServiceRepository extends EntityRepository<PipelineService>
       entity.setChangeDescription(changeDescription);
     }
 
+    @Override
+    public void setOwner(EntityReference owner) {
+      entity.setOwner(owner);
+    }
+
     @Override
     public void setDeleted(boolean flag) {
       entity.setDeleted(flag);

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/StorageServiceRepository.java

@@ -13,10 +13,12 @@
 
 package org.openmetadata.catalog.jdbi3;
 
+import static org.openmetadata.catalog.Entity.helper;
 import static org.openmetadata.catalog.util.EntityUtil.Fields;
 
 import java.io.IOException;
 import java.net.URI;
+import java.text.ParseException;
 import java.util.UUID;
 import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.entity.services.StorageService;
@@ -26,6 +28,8 @@ import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.util.EntityInterface;
 
 public class StorageServiceRepository extends EntityRepository<StorageService> {
+  private static final Fields UPDATE_FIELDS = new Fields(StorageServiceResource.FIELD_LIST, "owner");
+
   public StorageServiceRepository(CollectionDAO dao) {
     super(
         StorageServiceResource.COLLECTION_PATH,
@@ -34,14 +38,15 @@ public class StorageServiceRepository extends EntityRepository<StorageService> {
         dao.storageServiceDAO(),
         dao,
         Fields.EMPTY_FIELDS,
-        Fields.EMPTY_FIELDS,
+        UPDATE_FIELDS,
-        false,
         false,
+        true,
         false);
   }
 
   @Override
-  public StorageService setFields(StorageService entity, Fields fields) {
+  public StorageService setFields(StorageService entity, Fields fields) throws IOException, ParseException {
+    entity.setOwner(fields.contains("owner") ? getOwner(entity) : null);
     return entity;
   }
 
@@ -56,18 +61,29 @@ public class StorageServiceRepository extends EntityRepository<StorageService> {
   }
 
   @Override
-  public void prepare(StorageService entity) {
+  public void prepare(StorageService entity) throws IOException, ParseException {
-    /* Nothing to do */
+    // Check if owner is valid and set the relationship
+    entity.setOwner(helper(entity).validateOwnerOrNull());
   }
 
   @Override
   public void storeEntity(StorageService service, boolean update) throws IOException {
+    // Relationships and fields such as href are derived and not stored as part of json
+    EntityReference owner = service.getOwner();
+
+    // Don't store owner, database, href and tags as JSON. Build it on the fly based on relationships
+    service.withOwner(null).withHref(null);
+
     store(service.getId(), service, update);
+
+    // Restore the relationships
+    service.withOwner(owner);
   }
 
   @Override
   public void storeRelationships(StorageService entity) {
-    /* Nothing to do */
+    // Add owner relationship
+    setOwner(entity, entity.getOwner());
   }
 
   public static class StorageServiceEntityInterface implements EntityInterface<StorageService> {
@@ -112,6 +128,11 @@ public class StorageServiceRepository extends EntityRepository<StorageService> {
       return entity.getUpdatedBy();
     }
 
+    @Override
+    public EntityReference getOwner() {
+      return entity.getOwner();
+    }
+
     @Override
     public long getUpdatedAt() {
       return entity.getUpdatedAt();
@@ -169,6 +190,11 @@ public class StorageServiceRepository extends EntityRepository<StorageService> {
       entity.setChangeDescription(changeDescription);
     }
 
+    @Override
+    public void setOwner(EntityReference owner) {
+      entity.setOwner(owner);
+    }
+
     @Override
     public void setDeleted(boolean flag) {
       entity.setDeleted(flag);

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/databases/DatabaseResource.java

@@ -359,7 +359,7 @@ public class DatabaseResource {
   @Operation(
       summary = "Create or update database",
       tags = "databases",
-      description = "Create a database, it it does not exist or update an existing database.",
+      description = "Create a database, if it does not exist or update an existing database.",
       responses = {
         @ApiResponse(
             responseCode = "200",

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

@@ -13,6 +13,8 @@
 
 package org.openmetadata.catalog.resources.services.dashboard;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import io.swagger.annotations.Api;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -23,8 +25,11 @@ import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.security.GeneralSecurityException;
 import java.text.ParseException;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -44,15 +49,19 @@ import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.api.services.CreateDashboardService;
 import org.openmetadata.catalog.entity.services.DashboardService;
+import org.openmetadata.catalog.exception.EntityNotFoundException;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.DashboardServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.type.EntityHistory;
+import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.RestUtil.DeleteResponse;
 import org.openmetadata.catalog.util.RestUtil.PutResponse;
@@ -68,6 +77,20 @@ public class DashboardServiceResource {
   private final DashboardServiceRepository dao;
   private final Authorizer authorizer;
 
+  static final String FIELDS = "owner";
+  public static final List<String> FIELD_LIST = Arrays.asList(FIELDS.replace(" ", "").split(","));
+
+  public static ResultList<DashboardService> addHref(UriInfo uriInfo, ResultList<DashboardService> services) {
+    Optional.ofNullable(services.getData()).orElse(Collections.emptyList()).forEach(i -> addHref(uriInfo, i));
+    return services;
+  }
+
+  public static DashboardService addHref(UriInfo uriInfo, DashboardService service) {
+    service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
+    Entity.withHref(uriInfo, service.getOwner());
+    return service;
+  }
+
   public DashboardServiceResource(CollectionDAO dao, Authorizer authorizer) {
     Objects.requireNonNull(dao, "DashboardServiceRepository must not be null");
     this.dao = new DashboardServiceRepository(dao);
@@ -99,6 +122,11 @@ public class DashboardServiceResource {
   public ResultList<DashboardService> list(
       @Context UriInfo uriInfo,
       @QueryParam("name") String name,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @DefaultValue("10") @Min(1) @Max(1000000) @QueryParam("limit") int limitParam,
       @Parameter(
               description = "Returns list of dashboard services before this cursor",
@@ -118,12 +146,15 @@ public class DashboardServiceResource {
           Include include)
       throws IOException, GeneralSecurityException, ParseException {
     RestUtil.validateCursors(before, after);
-
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    ResultList<DashboardService> services;
     if (before != null) { // Reverse paging
-      return dao.listBefore(uriInfo, null, null, limitParam, before, include);
+      services = dao.listBefore(uriInfo, fields, null, limitParam, before, include);
+    } else {
+      // Forward paging
+      services = dao.listAfter(uriInfo, fields, null, limitParam, after, include);
     }
-    // Forward paging
+    return addHref(uriInfo, services);
-    return dao.listAfter(uriInfo, null, null, limitParam, after, include);
   }
 
   @GET
@@ -144,6 +175,11 @@ public class DashboardServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("id") String id,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -151,7 +187,8 @@ public class DashboardServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.get(uriInfo, id, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.get(uriInfo, id, fields, include));
   }
 
   @GET
@@ -172,6 +209,11 @@ public class DashboardServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("name") String name,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -179,7 +221,8 @@ public class DashboardServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.getByName(uriInfo, name, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.getByName(uriInfo, name, fields, include));
   }
 
   @GET
@@ -251,7 +294,7 @@ public class DashboardServiceResource {
       throws IOException, ParseException {
     SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
     DashboardService service = getService(create, securityContext);
-    dao.create(uriInfo, service);
+    service = addHref(uriInfo, dao.create(uriInfo, service));
     return Response.created(service.getHref()).entity(service).build();
   }
 
@@ -273,9 +316,20 @@ public class DashboardServiceResource {
   public Response update(
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateDashboardService update)
       throws IOException, ParseException {
-    SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, FIELDS);
-    DashboardService service = getService(update, securityContext);
+    EntityReference owner = null;
+    DashboardService service;
+    // Try to find the owner if entity exists
+    try {
+      service = dao.getByName(uriInfo, update.getName(), fields);
+      owner = helper(service).validateOwnerOrNull();
+    } catch (EntityNotFoundException e) {
+      // This is a create request if entity is not found. ignore exception
+    }
+    service = getService(update, securityContext);
+    SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext, owner);
     PutResponse<DashboardService> response = dao.createOrUpdate(uriInfo, service, true);
+    addHref(uriInfo, response.getEntity());
     return response.toResponse();
   }
 
@@ -314,6 +368,7 @@ public class DashboardServiceResource {
         .withDashboardUrl(create.getDashboardUrl())
         .withUsername(create.getUsername())
         .withPassword(create.getPassword())
+        .withOwner(create.getOwner())
         .withIngestionSchedule(create.getIngestionSchedule())
         .withUpdatedBy(securityContext.getUserPrincipal().getName())
         .withUpdatedAt(System.currentTimeMillis());

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

@@ -13,6 +13,8 @@
 
 package org.openmetadata.catalog.resources.services.database;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import io.swagger.annotations.Api;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -24,8 +26,10 @@ import java.io.UnsupportedEncodingException;
 import java.security.GeneralSecurityException;
 import java.text.ParseException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -45,14 +49,17 @@ import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.api.services.CreateDatabaseService;
 import org.openmetadata.catalog.entity.services.DatabaseService;
+import org.openmetadata.catalog.exception.EntityNotFoundException;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.DatabaseServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.type.EntityHistory;
+import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.Include;
 import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.RestUtil;
@@ -70,9 +77,20 @@ public class DatabaseServiceResource {
   private final DatabaseServiceRepository dao;
   private final Authorizer authorizer;
 
-  static final String FIELDS = "airflowPipeline";
+  static final String FIELDS = "airflowPipeline,owner";
   public static final List<String> FIELD_LIST = Arrays.asList(FIELDS.replace(" ", "").split(","));
 
+  public static ResultList<DatabaseService> addHref(UriInfo uriInfo, ResultList<DatabaseService> dbServices) {
+    Optional.ofNullable(dbServices.getData()).orElse(Collections.emptyList()).forEach(i -> addHref(uriInfo, i));
+    return dbServices;
+  }
+
+  public static DatabaseService addHref(UriInfo uriInfo, DatabaseService service) {
+    service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
+    Entity.withHref(uriInfo, service.getOwner());
+    return service;
+  }
+
   public DatabaseServiceResource(CollectionDAO dao, Authorizer authorizer) {
     Objects.requireNonNull(dao, "DatabaseServiceRepository must not be null");
     this.dao = new DatabaseServiceRepository(dao);
@@ -126,10 +144,13 @@ public class DatabaseServiceResource {
       throws IOException, GeneralSecurityException, ParseException {
     RestUtil.validateCursors(before, after);
     EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    ResultList<DatabaseService> dbServices;
     if (before != null) {
-      return dao.listBefore(uriInfo, fields, null, limitParam, before, include);
+      dbServices = dao.listBefore(uriInfo, fields, null, limitParam, before, include);
+    } else {
+      dbServices = dao.listAfter(uriInfo, fields, null, limitParam, after, include);
     }
-    return dao.listAfter(uriInfo, fields, null, limitParam, after, include);
+    return addHref(uriInfo, dbServices);
   }
 
   @GET
@@ -163,7 +184,7 @@ public class DatabaseServiceResource {
           Include include)
       throws IOException, ParseException {
     EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
-    return dao.get(uriInfo, id, fields, include);
+    return addHref(uriInfo, dao.get(uriInfo, id, fields, include));
   }
 
   @GET
@@ -197,7 +218,7 @@ public class DatabaseServiceResource {
           Include include)
       throws IOException, ParseException {
     EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
-    return dao.getByName(uriInfo, name, fields, include);
+    return addHref(uriInfo, dao.getByName(uriInfo, name, fields, include));
   }
 
   @GET
@@ -269,7 +290,7 @@ public class DatabaseServiceResource {
       throws IOException, ParseException {
     SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
     DatabaseService service = getService(create, securityContext);
-    dao.create(uriInfo, service);
+    service = addHref(uriInfo, dao.create(uriInfo, service));
     return Response.created(service.getHref()).entity(service).build();
   }
 
@@ -291,9 +312,20 @@ public class DatabaseServiceResource {
   public Response update(
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateDatabaseService update)
       throws IOException, ParseException {
-    SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, FIELDS);
-    DatabaseService service = getService(update, securityContext);
+    EntityReference owner = null;
+    DatabaseService service;
+    // Try to find the owner if entity exists
+    try {
+      service = dao.getByName(uriInfo, update.getName(), fields);
+      owner = helper(service).validateOwnerOrNull();
+    } catch (EntityNotFoundException e) {
+      // This is a create request if entity is not found. ignore exception
+    }
+    service = getService(update, securityContext);
+    SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext, owner);
     PutResponse<DatabaseService> response = dao.createOrUpdate(uriInfo, service, true);
+    addHref(uriInfo, response.getEntity());
     return response.toResponse();
   }
 
@@ -330,6 +362,7 @@ public class DatabaseServiceResource {
         .withDescription(create.getDescription())
         .withServiceType(create.getServiceType())
         .withDatabaseConnection(create.getDatabaseConnection())
+        .withOwner(create.getOwner())
         .withUpdatedBy(securityContext.getUserPrincipal().getName())
         .withUpdatedAt(System.currentTimeMillis());
   }

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

@@ -13,6 +13,8 @@
 
 package org.openmetadata.catalog.resources.services.messaging;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import io.swagger.annotations.Api;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -23,8 +25,11 @@ import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.security.GeneralSecurityException;
 import java.text.ParseException;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -44,15 +49,19 @@ import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.api.services.CreateMessagingService;
 import org.openmetadata.catalog.entity.services.MessagingService;
+import org.openmetadata.catalog.exception.EntityNotFoundException;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.MessagingServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.type.EntityHistory;
+import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.RestUtil.DeleteResponse;
 import org.openmetadata.catalog.util.RestUtil.PutResponse;
@@ -68,6 +77,20 @@ public class MessagingServiceResource {
   private final MessagingServiceRepository dao;
   private final Authorizer authorizer;
 
+  static final String FIELDS = "owner";
+  public static final List<String> FIELD_LIST = Arrays.asList(FIELDS.replace(" ", "").split(","));
+
+  public static ResultList<MessagingService> addHref(UriInfo uriInfo, ResultList<MessagingService> services) {
+    Optional.ofNullable(services.getData()).orElse(Collections.emptyList()).forEach(i -> addHref(uriInfo, i));
+    return services;
+  }
+
+  public static MessagingService addHref(UriInfo uriInfo, MessagingService service) {
+    service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
+    Entity.withHref(uriInfo, service.getOwner());
+    return service;
+  }
+
   public MessagingServiceResource(CollectionDAO dao, Authorizer authorizer) {
     Objects.requireNonNull(dao, "MessagingServiceRepository must not be null");
     this.dao = new MessagingServiceRepository(dao);
@@ -101,6 +124,11 @@ public class MessagingServiceResource {
   public ResultList<MessagingService> list(
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(description = "Limit number services returned. (1 to 1000000, " + "default 10)")
           @DefaultValue("10")
           @Min(1)
@@ -121,11 +149,15 @@ public class MessagingServiceResource {
           Include include)
       throws IOException, ParseException, GeneralSecurityException {
     RestUtil.validateCursors(before, after);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    ResultList<MessagingService> services;
     if (before != null) { // Reverse paging
-      return dao.listBefore(uriInfo, null, null, limitParam, before, include);
+      services = dao.listBefore(uriInfo, fields, null, limitParam, before, include);
+    } else {
+      // Forward paging or first page
+      services = dao.listAfter(uriInfo, fields, null, limitParam, after, include);
     }
-    // Forward paging or first page
+    return addHref(uriInfo, services);
-    return dao.listAfter(uriInfo, null, null, limitParam, after, include);
   }
 
   @GET
@@ -146,6 +178,11 @@ public class MessagingServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("id") String id,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -153,7 +190,8 @@ public class MessagingServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.get(uriInfo, id, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.get(uriInfo, id, fields, include));
   }
 
   @GET
@@ -174,6 +212,11 @@ public class MessagingServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("name") String name,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -181,7 +224,8 @@ public class MessagingServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.getByName(uriInfo, name, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.getByName(uriInfo, name, fields, include));
   }
 
   @GET
@@ -253,7 +297,7 @@ public class MessagingServiceResource {
       throws IOException, ParseException {
     SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
     MessagingService service = getService(create, securityContext);
-    dao.create(uriInfo, service);
+    service = addHref(uriInfo, dao.create(uriInfo, service));
     return Response.created(service.getHref()).entity(service).build();
   }
 
@@ -279,9 +323,20 @@ public class MessagingServiceResource {
           String id,
       @Valid CreateMessagingService update)
       throws IOException, ParseException {
-    SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, FIELDS);
-    MessagingService service = getService(update, securityContext);
+    EntityReference owner = null;
+    MessagingService service;
+    // Try to find the owner if entity exists
+    try {
+      service = dao.getByName(uriInfo, update.getName(), fields);
+      owner = helper(service).validateOwnerOrNull();
+    } catch (EntityNotFoundException e) {
+      // This is a create request if entity is not found. ignore exception
+    }
+    service = getService(update, securityContext);
+    SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext, owner);
     PutResponse<MessagingService> response = dao.createOrUpdate(uriInfo, service, true);
+    addHref(uriInfo, response.getEntity());
     return response.toResponse();
   }
 
@@ -319,6 +374,7 @@ public class MessagingServiceResource {
         .withBrokers(create.getBrokers())
         .withSchemaRegistry(create.getSchemaRegistry())
         .withIngestionSchedule(create.getIngestionSchedule())
+        .withOwner(create.getOwner())
         .withUpdatedBy(securityContext.getUserPrincipal().getName())
         .withUpdatedAt(System.currentTimeMillis());
   }

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

@@ -13,6 +13,8 @@
 
 package org.openmetadata.catalog.resources.services.pipeline;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import io.swagger.annotations.Api;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -23,8 +25,11 @@ import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.security.GeneralSecurityException;
 import java.text.ParseException;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -44,8 +49,10 @@ import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.api.services.CreatePipelineService;
 import org.openmetadata.catalog.entity.services.PipelineService;
+import org.openmetadata.catalog.exception.EntityNotFoundException;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.PipelineServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
@@ -54,6 +61,7 @@ import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.type.EntityHistory;
 import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.RestUtil.DeleteResponse;
 import org.openmetadata.catalog.util.RestUtil.PutResponse;
@@ -69,8 +77,18 @@ public class PipelineServiceResource {
   private final PipelineServiceRepository dao;
   private final Authorizer authorizer;
 
-  public static EntityReference addHref(UriInfo uriInfo, EntityReference service) {
+  static final String FIELDS = "owner";
-    return service.withHref(RestUtil.getHref(uriInfo, "v1/services/pipelineServices/", service.getId()));
+  public static final List<String> FIELD_LIST = Arrays.asList(FIELDS.replace(" ", "").split(","));
+
+  public static ResultList<PipelineService> addHref(UriInfo uriInfo, ResultList<PipelineService> services) {
+    Optional.ofNullable(services.getData()).orElse(Collections.emptyList()).forEach(i -> addHref(uriInfo, i));
+    return services;
+  }
+
+  public static PipelineService addHref(UriInfo uriInfo, PipelineService service) {
+    service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
+    Entity.withHref(uriInfo, service.getOwner());
+    return service;
   }
 
   public PipelineServiceResource(CollectionDAO dao, Authorizer authorizer) {
@@ -106,6 +124,11 @@ public class PipelineServiceResource {
   public ResultList<PipelineService> list(
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(description = "Limit number services returned. (1 to 1000000, " + "default 10)")
           @DefaultValue("10")
           @Min(1)
@@ -126,12 +149,15 @@ public class PipelineServiceResource {
           Include include)
       throws IOException, GeneralSecurityException, ParseException {
     RestUtil.validateCursors(before, after);
-
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    ResultList<PipelineService> services;
     if (before != null) { // Reverse paging
-      return dao.listBefore(uriInfo, null, null, limitParam, before, include);
+      services = dao.listBefore(uriInfo, fields, null, limitParam, before, include);
+    } else {
+      // Forward paging or first page
+      services = dao.listAfter(uriInfo, fields, null, limitParam, after, include);
     }
-    // Forward paging or first page
+    return addHref(uriInfo, services);
-    return dao.listAfter(uriInfo, null, null, limitParam, after, include);
   }
 
   @GET
@@ -152,6 +178,11 @@ public class PipelineServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("id") String id,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -159,7 +190,8 @@ public class PipelineServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.get(uriInfo, id, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.get(uriInfo, id, fields, include));
   }
 
   @GET
@@ -180,6 +212,11 @@ public class PipelineServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("name") String name,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -187,7 +224,8 @@ public class PipelineServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.getByName(uriInfo, name, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.getByName(uriInfo, name, fields, include));
   }
 
   @GET
@@ -259,7 +297,7 @@ public class PipelineServiceResource {
       throws IOException, ParseException {
     SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
     PipelineService service = getService(create, securityContext);
-    dao.create(uriInfo, service);
+    service = addHref(uriInfo, dao.create(uriInfo, service));
     return Response.created(service.getHref()).entity(service).build();
   }
 
@@ -281,9 +319,20 @@ public class PipelineServiceResource {
   public Response update(
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreatePipelineService update)
       throws IOException, ParseException {
-    SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, FIELDS);
-    PipelineService service = getService(update, securityContext);
+    EntityReference owner = null;
+    PipelineService service;
+    // Try to find the owner if entity exists
+    try {
+      service = dao.getByName(uriInfo, update.getName(), fields);
+      owner = helper(service).validateOwnerOrNull();
+    } catch (EntityNotFoundException e) {
+      // This is a create request if entity is not found. ignore exception
+    }
+    service = getService(update, securityContext);
+    SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext, owner);
     PutResponse<PipelineService> response = dao.createOrUpdate(uriInfo, service, true);
+    addHref(uriInfo, response.getEntity());
     return response.toResponse();
   }
 
@@ -321,6 +370,7 @@ public class PipelineServiceResource {
         .withServiceType(create.getServiceType())
         .withPipelineUrl(create.getPipelineUrl())
         .withIngestionSchedule(create.getIngestionSchedule())
+        .withOwner(create.getOwner())
         .withUpdatedBy(securityContext.getUserPrincipal().getName())
         .withUpdatedAt(System.currentTimeMillis());
   }

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

@@ -13,6 +13,8 @@
 
 package org.openmetadata.catalog.resources.services.storage;
 
+import static org.openmetadata.catalog.Entity.helper;
+
 import io.swagger.annotations.Api;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -23,8 +25,11 @@ import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.security.GeneralSecurityException;
 import java.text.ParseException;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
@@ -44,15 +49,19 @@ import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
+import org.openmetadata.catalog.Entity;
 import org.openmetadata.catalog.api.services.CreateStorageService;
 import org.openmetadata.catalog.entity.services.StorageService;
+import org.openmetadata.catalog.exception.EntityNotFoundException;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.StorageServiceRepository;
 import org.openmetadata.catalog.resources.Collection;
 import org.openmetadata.catalog.security.Authorizer;
 import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.type.EntityHistory;
+import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.Include;
+import org.openmetadata.catalog.util.EntityUtil;
 import org.openmetadata.catalog.util.RestUtil;
 import org.openmetadata.catalog.util.RestUtil.DeleteResponse;
 import org.openmetadata.catalog.util.RestUtil.PutResponse;
@@ -68,6 +77,20 @@ public class StorageServiceResource {
   private final StorageServiceRepository dao;
   private final Authorizer authorizer;
 
+  static final String FIELDS = "owner";
+  public static final List<String> FIELD_LIST = Arrays.asList(FIELDS.replace(" ", "").split(","));
+
+  public static ResultList<StorageService> addHref(UriInfo uriInfo, ResultList<StorageService> services) {
+    Optional.ofNullable(services.getData()).orElse(Collections.emptyList()).forEach(i -> addHref(uriInfo, i));
+    return services;
+  }
+
+  public static StorageService addHref(UriInfo uriInfo, StorageService service) {
+    service.setHref(RestUtil.getHref(uriInfo, COLLECTION_PATH, service.getId()));
+    Entity.withHref(uriInfo, service.getOwner());
+    return service;
+  }
+
   public StorageServiceResource(CollectionDAO dao, Authorizer authorizer) {
     Objects.requireNonNull(dao, "StorageServiceRepository must not be null");
     this.dao = new StorageServiceRepository(dao);
@@ -101,7 +124,12 @@ public class StorageServiceResource {
   public ResultList<StorageService> list(
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
-      @Parameter(description = "Limit number services returned. (1 to 1000000, " + "default 10)")
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
+      @Parameter(description = "Limit number of services returned. (1 to 1000000, " + "default 10)")
           @DefaultValue("10")
           @Min(1)
           @Max(1000000)
@@ -121,11 +149,15 @@ public class StorageServiceResource {
           Include include)
       throws IOException, GeneralSecurityException, ParseException {
     RestUtil.validateCursors(before, after);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    ResultList<StorageService> services;
     if (before != null) { // Reverse paging
-      return dao.listBefore(uriInfo, null, null, limitParam, before, include);
+      services = dao.listBefore(uriInfo, fields, null, limitParam, before, include);
+    } else {
+      // Forward paging or first page
+      services = dao.listAfter(uriInfo, fields, null, limitParam, after, include);
     }
-    // Forward paging or first page
+    return addHref(uriInfo, services);
-    return dao.listAfter(uriInfo, null, null, limitParam, after, include);
   }
 
   @GET
@@ -146,6 +178,11 @@ public class StorageServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("id") String id,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -153,7 +190,8 @@ public class StorageServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.get(uriInfo, id, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.get(uriInfo, id, fields, include));
   }
 
   @GET
@@ -174,6 +212,11 @@ public class StorageServiceResource {
       @Context UriInfo uriInfo,
       @Context SecurityContext securityContext,
       @PathParam("name") String name,
+      @Parameter(
+              description = "Fields requested in the returned resource",
+              schema = @Schema(type = "string", example = FIELDS))
+          @QueryParam("fields")
+          String fieldsParam,
       @Parameter(
               description = "Include all, deleted, or non-deleted entities.",
               schema = @Schema(implementation = Include.class))
@@ -181,7 +224,8 @@ public class StorageServiceResource {
           @DefaultValue("non-deleted")
           Include include)
       throws IOException, ParseException {
-    return dao.getByName(uriInfo, name, null, include);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, fieldsParam);
+    return addHref(uriInfo, dao.getByName(uriInfo, name, fields, include));
   }
 
   @GET
@@ -250,9 +294,9 @@ public class StorageServiceResource {
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateStorageService create)
       throws IOException, ParseException {
     SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
-    StorageService databaseService = getService(create, securityContext);
+    StorageService service = getService(create, securityContext);
-    dao.create(uriInfo, databaseService);
+    service = addHref(uriInfo, dao.create(uriInfo, service));
-    return Response.created(databaseService.getHref()).entity(databaseService).build();
+    return Response.created(service.getHref()).entity(service).build();
   }
 
   @PUT
@@ -271,9 +315,20 @@ public class StorageServiceResource {
   public Response update(
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateStorageService update)
       throws IOException, ParseException {
-    SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+    EntityUtil.Fields fields = new EntityUtil.Fields(FIELD_LIST, FIELDS);
-    StorageService databaseService = getService(update, securityContext);
+    EntityReference owner = null;
-    PutResponse<StorageService> response = dao.createOrUpdate(uriInfo, databaseService);
+    StorageService service;
+    // Try to find the owner if entity exists
+    try {
+      service = dao.getByName(uriInfo, update.getName(), fields);
+      owner = helper(service).validateOwnerOrNull();
+    } catch (EntityNotFoundException e) {
+      // This is a create request if entity is not found. ignore exception
+    }
+    service = getService(update, securityContext);
+    SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext, owner);
+    PutResponse<StorageService> response = dao.createOrUpdate(uriInfo, service, true);
+    addHref(uriInfo, response.getEntity());
     return response.toResponse();
   }
 
@@ -308,6 +363,7 @@ public class StorageServiceResource {
         .withName(create.getName())
         .withDescription(create.getDescription())
         .withServiceType(create.getServiceType())
+        .withOwner(create.getOwner())
         .withUpdatedBy(securityContext.getUserPrincipal().getName())
         .withUpdatedAt(System.currentTimeMillis());
   }

catalog-rest-service/src/main/resources/json/schema/api/data/createChart.json

@@ -40,11 +40,11 @@
       "default": null
     },
     "owner": {
-      "description": "Owner of this database",
+      "description": "Owner of this chart",
       "$ref": "../../type/entityReference.json"
     },
     "service": {
-      "description": "Link to the database service where this database is hosted in",
+      "description": "Link to the chart service where this chart is hosted in",
       "$ref": "../../type/entityReference.json"
     }
   },

catalog-rest-service/src/main/resources/json/schema/api/data/createDashboard.json

@@ -1,5 +1,5 @@
 {
-  "$id": "https://open-metadata.org/schema/api/data/createDatabase.json",
+  "$id": "https://open-metadata.org/schema/api/data/createDashboard.json",
   "$schema": "http://json-schema.org/draft-07/schema#",
   "title": "CreateDashboardRequest",
   "description": "Create Dashboard entity request",
@@ -33,7 +33,7 @@
       "default": null
     },
     "tags": {
-      "description": "Tags for this chart",
+      "description": "Tags for this dashboard",
       "type": "array",
       "items": {
         "$ref": "../../type/tagLabel.json"
@@ -41,11 +41,11 @@
       "default": null
     },
     "owner": {
-      "description": "Owner of this database",
+      "description": "Owner of this dashboard",
       "$ref": "../../type/entityReference.json"
     },
     "service": {
-      "description": "Link to the database service where this database is hosted in",
+      "description": "Link to the dashboard service where this dashboard is hosted in",
       "$ref": "../../type/entityReference.json"
     }
   },

catalog-rest-service/src/main/resources/json/schema/api/data/createPipeline.json

@@ -17,7 +17,7 @@
       "type": "string"
     },
     "description": {
-      "description": "Description of the database instance. What it has and how to use it.",
+      "description": "Description of the pipeline instance. What it has and how to use it.",
       "type": "string"
     },
     "pipelineUrl": {
@@ -54,11 +54,11 @@
       "default": null
     },
     "owner": {
-      "description": "Owner of this database",
+      "description": "Owner of this pipeline",
       "$ref": "../../type/entityReference.json"
     },
     "service": {
-      "description": "Link to the database service where this database is hosted in",
+      "description": "Link to the pipeline service where this pipeline is hosted in",
       "$ref": "../../type/entityReference.json"
     }
   },

catalog-rest-service/src/main/resources/json/schema/api/services/createDashboardService.json

@@ -34,6 +34,10 @@
     "ingestionSchedule": {
       "description": "Schedule for running metadata ingestion jobs",
       "$ref": "../../type/schedule.json"
+    },
+    "owner": {
+      "description": "Owner of this dashboard service.",
+      "$ref": "../../type/entityReference.json"
     }
   },
   "required": ["name", "serviceType", "dashboardUrl"]

catalog-rest-service/src/main/resources/json/schema/api/services/createDatabaseService.json

@@ -21,6 +21,10 @@
     },
     "databaseConnection": {
       "$ref": "../../entity/services/databaseService.json#/definitions/databaseConnection"
+    },
+    "owner": {
+      "description": "Owner of this database service.",
+      "$ref": "../../type/entityReference.json"
     }
   },
   "required": ["name", "serviceType", "databaseConnection"]

catalog-rest-service/src/main/resources/json/schema/api/services/createMessagingService.json

@@ -30,6 +30,10 @@
     "ingestionSchedule": {
       "description": "Schedule for running metadata ingestion jobs",
       "$ref": "../../type/schedule.json"
+    },
+    "owner": {
+      "description": "Owner of this messaging service.",
+      "$ref": "../../type/entityReference.json"
     }
   },
   "required": ["name", "serviceType", "brokers"]

catalog-rest-service/src/main/resources/json/schema/api/services/createPipelineService.json

@@ -26,6 +26,10 @@
     "ingestionSchedule": {
       "description": "Schedule for running pipeline ingestion jobs",
       "$ref": "../../type/schedule.json"
+    },
+    "owner": {
+      "description": "Owner of this pipeline service.",
+      "$ref": "../../type/entityReference.json"
     }
   },
   "required": ["name", "serviceType", "pipelineUrl"]

catalog-rest-service/src/main/resources/json/schema/api/services/createStorageService.json

@@ -17,6 +17,10 @@
     },
     "serviceType": {
       "$ref": "../../type/storage.json#/definitions/storageServiceType"
+    },
+    "owner": {
+      "description": "Owner of this storage service.",
+      "$ref": "../../type/entityReference.json"
     }
   },
   "required": ["name"]

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

@@ -63,6 +63,10 @@
       "description": "User who made the update.",
       "type": "string"
     },
+    "owner": {
+      "description": "Owner of this dashboard service.",
+      "$ref": "../../type/entityReference.json"
+    },
     "dashboardUrl": {
       "description": "Dashboard Service URL. This will be used to make REST API calls to Dashboard Service.",
       "type": "string",

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

@@ -148,6 +148,10 @@
       "description": "User who made the update.",
       "type": "string"
     },
+    "owner": {
+      "description": "Owner of this database service.",
+      "$ref": "../../type/entityReference.json"
+    },
     "href": {
       "description": "Link to the resource corresponding to this database service.",
       "$ref": "../../type/basic.json#/definitions/href"

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

@@ -75,6 +75,10 @@
       "description": "Schedule for running metadata ingestion jobs.",
       "$ref": "../../type/schedule.json"
     },
+    "owner": {
+      "description": "Owner of this messaging service.",
+      "$ref": "../../type/entityReference.json"
+    },
     "href": {
       "description": "Link to the resource corresponding to this messaging service.",
       "$ref": "../../type/basic.json#/definitions/href"

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

@@ -66,6 +66,10 @@
       "description": "Schedule for running metadata ingestion jobs.",
       "$ref": "../../type/schedule.json"
     },
+    "owner": {
+      "description": "Owner of this pipeline service.",
+      "$ref": "../../type/entityReference.json"
+    },
     "href": {
       "description": "Link to the resource corresponding to this pipeline service.",
       "$ref": "../../type/basic.json#/definitions/href"

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

@@ -44,6 +44,10 @@
       "description": "Link to the resource corresponding to this storage service.",
       "$ref": "../../type/basic.json#/definitions/href"
     },
+    "owner": {
+      "description": "Owner of this storage service.",
+      "$ref": "../../type/entityReference.json"
+    },
     "changeDescription": {
       "description": "Change that lead to this version of the entity.",
       "$ref": "../../type/entityHistory.json#/definitions/changeDescription"

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

@@ -863,7 +863,12 @@ public abstract class EntityResourceTest<T, K> extends CatalogApplicationTest {
     entityInterface = getEntityInterface(entity);
     // For service resources, we allow update of non-empty description via PUT
     List<Class<?>> services =
-        Arrays.asList(DatabaseService.class, PipelineService.class, DashboardService.class, MessagingService.class);
+        Arrays.asList(
+            DatabaseService.class,
+            PipelineService.class,
+            StorageService.class,
+            DashboardService.class,
+            MessagingService.class);
     if (services.contains(entity.getClass())) {
       assertNotEquals(oldVersion, entityInterface.getVersion()); // Version did change
       assertEquals("updatedDescription", entityInterface.getDescription()); // Description did change

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

@@ -54,9 +54,9 @@ public class DashboardServiceResourceTest extends EntityResourceTest<DashboardSe
         DashboardService.class,
         DashboardServiceList.class,
         "services/dashboardServices",
-        "",
+        "owner",
-        false,
         false,
+        true,
         false,
         false);
     this.supportsPatch = false;
@@ -71,7 +71,7 @@ public class DashboardServiceResourceTest extends EntityResourceTest<DashboardSe
             () -> createEntity(createRequest(test).withServiceType(null), ADMIN_AUTH_HEADERS));
     TestUtils.assertResponse(exception, BAD_REQUEST, "[serviceType must not be null]");
 
-    // Create dashboard with mandatory brokers field empty
+    // Create dashboard with mandatory dashboardUrl field empty
     exception =
         assertThrows(
             HttpResponseException.class,
@@ -192,16 +192,18 @@ public class DashboardServiceResourceTest extends EntityResourceTest<DashboardSe
   }
 
   @Test
-  void put_update_as_non_admin_401(TestInfo test) throws IOException {
+  void put_update_as_non_owner_401(TestInfo test) throws IOException {
-    Map<String, String> authHeaders = ADMIN_AUTH_HEADERS;
+    createAndCheckEntity(
-    createAndCheckEntity(createRequest(test).withDescription(null).withIngestionSchedule(null), authHeaders);
+        createRequest(test).withDescription(null).withIngestionSchedule(null).withOwner(USER_OWNER1),
+        ADMIN_AUTH_HEADERS);
 
     // Update dashboard description and ingestion service that are null
     HttpResponseException exception =
         assertThrows(
             HttpResponseException.class,
             () -> updateAndCheckEntity(createRequest(test), OK, TEST_AUTH_HEADERS, UpdateType.NO_CHANGE, null));
-    TestUtils.assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "is not admin");
+    TestUtils.assertResponse(
+        exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "does not have permissions");
   }
 
   @Override
@@ -214,6 +216,7 @@ public class DashboardServiceResourceTest extends EntityResourceTest<DashboardSe
           .withDashboardUrl(new URI("http://192.1.1.1:0"))
           //          .withIngestionSchedule(new Schedule().withStartDate(new Date()).withRepeatFrequency("P1D"))
           .withIngestionSchedule(null)
+          .withOwner(owner)
           .withDescription(description);
     } catch (URISyntaxException e) {
       e.printStackTrace();
@@ -225,7 +228,10 @@ public class DashboardServiceResourceTest extends EntityResourceTest<DashboardSe
   public void validateCreatedEntity(
       DashboardService service, CreateDashboardService createRequest, Map<String, String> authHeaders) {
     validateCommonEntityFields(
-        getEntityInterface(service), createRequest.getDescription(), getPrincipal(authHeaders), null);
+        getEntityInterface(service),
+        createRequest.getDescription(),
+        getPrincipal(authHeaders),
+        createRequest.getOwner());
     assertEquals(createRequest.getName(), service.getName());
 
     Schedule expectedIngestion = createRequest.getIngestionSchedule();
@@ -253,14 +259,14 @@ public class DashboardServiceResourceTest extends EntityResourceTest<DashboardSe
 
   @Override
   public void validateGetWithDifferentFields(DashboardService service, boolean byName) throws HttpResponseException {
-    // No fields support
+    String fields = "owner";
-    String fields = "";
     service =
         byName
             ? getEntityByName(service.getName(), fields, ADMIN_AUTH_HEADERS)
             : getEntity(service.getId(), fields, ADMIN_AUTH_HEADERS);
     TestUtils.assertListNotNull(
         service.getHref(),
+        service.getOwner(),
         service.getVersion(),
         service.getUpdatedBy(),
         service.getServiceType(),

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

@@ -64,9 +64,9 @@ public class DatabaseServiceResourceTest extends EntityResourceTest<DatabaseServ
         DatabaseService.class,
         DatabaseServiceList.class,
         "services/databaseServices",
-        "",
+        "owner",
-        false,
         false,
+        true,
         false,
         false);
     this.supportsPatch = false;
@@ -175,16 +175,16 @@ public class DatabaseServiceResourceTest extends EntityResourceTest<DatabaseServ
   }
 
   @Test
-  void put_update_as_non_admin_401(TestInfo test) throws IOException {
+  void put_update_as_non_owner_401(TestInfo test) throws IOException {
-    Map<String, String> authHeaders = ADMIN_AUTH_HEADERS;
+    createAndCheckEntity(createRequest(test).withDescription(null).withOwner(USER_OWNER1), ADMIN_AUTH_HEADERS);
-    createAndCheckEntity(createRequest(test).withDescription(null), authHeaders);
 
-    // Update as non admin should be forbidden
+    // Update as non owner should be forbidden
     HttpResponseException exception =
         assertThrows(
             HttpResponseException.class,
             () -> updateAndCheckEntity(createRequest(test), OK, TEST_AUTH_HEADERS, UpdateType.MINOR_UPDATE, null));
-    TestUtils.assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "is not admin");
+    TestUtils.assertResponse(
+        exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "does not have permissions");
   }
 
   @Override
@@ -194,6 +194,7 @@ public class DatabaseServiceResourceTest extends EntityResourceTest<DatabaseServ
         .withName(name)
         .withServiceType(DatabaseServiceType.Snowflake)
         .withDatabaseConnection(TestUtils.DATABASE_CONNECTION)
+        .withOwner(owner)
         .withDescription(description);
   }
 
@@ -201,7 +202,10 @@ public class DatabaseServiceResourceTest extends EntityResourceTest<DatabaseServ
   public void validateCreatedEntity(
       DatabaseService service, CreateDatabaseService createRequest, Map<String, String> authHeaders) {
     validateCommonEntityFields(
-        getEntityInterface(service), createRequest.getDescription(), getPrincipal(authHeaders), null);
+        getEntityInterface(service),
+        createRequest.getDescription(),
+        getPrincipal(authHeaders),
+        createRequest.getOwner());
     assertEquals(createRequest.getName(), service.getName());
 
     // Validate Database Connection
@@ -224,16 +228,20 @@ public class DatabaseServiceResourceTest extends EntityResourceTest<DatabaseServ
     return new DatabaseServiceEntityInterface(entity);
   }
 
+  /**
+   * Validate returned fields GET .../databaseServices/{id}?fields="..." or GET
+   * .../databaseServices/name/{fqn}?fields="..."
+   */
   @Override
   public void validateGetWithDifferentFields(DatabaseService service, boolean byName) throws HttpResponseException {
-    // No fields support
+    String fields = "owner";
-    String fields = "";
     service =
         byName
             ? getEntityByName(service.getName(), fields, ADMIN_AUTH_HEADERS)
             : getEntity(service.getId(), fields, ADMIN_AUTH_HEADERS);
     TestUtils.assertListNotNull(
         service.getHref(),
+        service.getOwner(),
         service.getVersion(),
         service.getUpdatedBy(),
         service.getServiceType(),

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

@@ -61,9 +61,9 @@ public class MessagingServiceResourceTest extends EntityResourceTest<MessagingSe
         MessagingService.class,
         MessagingServiceList.class,
         "services/messagingServices",
-        "",
+        "owner",
-        false,
         false,
+        true,
         false,
         false);
     supportsPatch = false;
@@ -218,15 +218,18 @@ public class MessagingServiceResourceTest extends EntityResourceTest<MessagingSe
   }
 
   @Test
-  void put_update_as_non_admin_401(TestInfo test) throws IOException {
+  void put_update_as_non_owner_401(TestInfo test) throws IOException {
-    createAndCheckEntity(createRequest(test).withDescription(null).withIngestionSchedule(null), ADMIN_AUTH_HEADERS);
+    createAndCheckEntity(
+        createRequest(test).withDescription(null).withIngestionSchedule(null).withOwner(USER_OWNER1),
+        ADMIN_AUTH_HEADERS);
 
-    // Update messaging description as non admin and expect exception
+    // Update messaging description as non owner and expect exception
     HttpResponseException exception =
         assertThrows(
             HttpResponseException.class,
             () -> updateAndCheckEntity(createRequest(test), OK, TEST_AUTH_HEADERS, UpdateType.NO_CHANGE, null));
-    TestUtils.assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "is not admin");
+    TestUtils.assertResponse(
+        exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "does not have permissions");
   }
 
   @Override
@@ -239,6 +242,7 @@ public class MessagingServiceResourceTest extends EntityResourceTest<MessagingSe
         .withSchemaRegistry(SCHEMA_REGISTRY_URL)
         //            .withIngestionSchedule(new Schedule().withStartDate(new Date()).withRepeatFrequency("P1D"));
         .withDescription(description)
+        .withOwner(owner)
         .withIngestionSchedule(null);
   }
 
@@ -246,7 +250,10 @@ public class MessagingServiceResourceTest extends EntityResourceTest<MessagingSe
   public void validateCreatedEntity(
       MessagingService service, CreateMessagingService createRequest, Map<String, String> authHeaders) {
     validateCommonEntityFields(
-        getEntityInterface(service), createRequest.getDescription(), TestUtils.getPrincipal(authHeaders), null);
+        getEntityInterface(service),
+        createRequest.getDescription(),
+        TestUtils.getPrincipal(authHeaders),
+        createRequest.getOwner());
     Schedule expectedIngestion = createRequest.getIngestionSchedule();
     if (expectedIngestion != null) {
       assertEquals(expectedIngestion.getStartDate(), service.getIngestionSchedule().getStartDate());
@@ -274,14 +281,14 @@ public class MessagingServiceResourceTest extends EntityResourceTest<MessagingSe
 
   @Override
   public void validateGetWithDifferentFields(MessagingService service, boolean byName) throws HttpResponseException {
-    // No fields support
+    String fields = "owner";
-    String fields = "";
     service =
         byName
             ? getEntityByName(service.getName(), fields, ADMIN_AUTH_HEADERS)
             : getEntity(service.getId(), fields, ADMIN_AUTH_HEADERS);
     TestUtils.assertListNotNull(
         service.getHref(),
+        service.getOwner(),
         service.getVersion(),
         service.getUpdatedBy(),
         service.getServiceType(),

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

@@ -58,9 +58,9 @@ public class PipelineServiceResourceTest extends EntityResourceTest<PipelineServ
         PipelineService.class,
         PipelineServiceList.class,
         "services/pipelineServices",
-        "",
+        "owner",
-        false,
         false,
+        true,
         false,
         false);
     this.supportsPatch = false;
@@ -198,15 +198,18 @@ public class PipelineServiceResourceTest extends EntityResourceTest<PipelineServ
   }
 
   @Test
-  void put_update_as_non_admin_401(TestInfo test) throws IOException {
+  void put_update_as_non_owner_401(TestInfo test) throws IOException {
-    createAndCheckEntity(createRequest(test).withDescription(null).withIngestionSchedule(null), ADMIN_AUTH_HEADERS);
+    createAndCheckEntity(
+        createRequest(test).withDescription(null).withIngestionSchedule(null).withOwner(USER_OWNER1),
+        ADMIN_AUTH_HEADERS);
 
     // Update pipeline description and ingestion service that are null
     HttpResponseException exception =
         assertThrows(
             HttpResponseException.class,
             () -> updateAndCheckEntity(createRequest(test), OK, TEST_AUTH_HEADERS, UpdateType.NO_CHANGE, null));
-    TestUtils.assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "is not admin");
+    TestUtils.assertResponse(
+        exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "does not have permissions");
   }
 
   @Override
@@ -217,6 +220,7 @@ public class PipelineServiceResourceTest extends EntityResourceTest<PipelineServ
         .withServiceType(CreatePipelineService.PipelineServiceType.Airflow)
         .withPipelineUrl(PIPELINE_SERVICE_URL)
         .withDescription(description)
+        .withOwner(owner)
         .withIngestionSchedule(null);
   }
 
@@ -224,7 +228,10 @@ public class PipelineServiceResourceTest extends EntityResourceTest<PipelineServ
   public void validateCreatedEntity(
       PipelineService service, CreatePipelineService createRequest, Map<String, String> authHeaders) {
     validateCommonEntityFields(
-        getEntityInterface(service), createRequest.getDescription(), getPrincipal(authHeaders), null);
+        getEntityInterface(service),
+        createRequest.getDescription(),
+        getPrincipal(authHeaders),
+        createRequest.getOwner());
     assertEquals(createRequest.getName(), service.getName());
 
     Schedule expectedIngestion = createRequest.getIngestionSchedule();
@@ -253,14 +260,14 @@ public class PipelineServiceResourceTest extends EntityResourceTest<PipelineServ
 
   @Override
   public void validateGetWithDifferentFields(PipelineService service, boolean byName) throws HttpResponseException {
-    // No fields support
+    String fields = "owner";
-    String fields = "";
     service =
         byName
             ? getEntityByName(service.getName(), fields, ADMIN_AUTH_HEADERS)
             : getEntity(service.getId(), fields, ADMIN_AUTH_HEADERS);
     TestUtils.assertListNotNull(
         service.getHref(),
+        service.getOwner(),
         service.getVersion(),
         service.getUpdatedBy(),
         service.getServiceType(),

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

@@ -47,9 +47,9 @@ public class StorageServiceResourceTest extends EntityResourceTest<StorageServic
         StorageService.class,
         StorageServiceList.class,
         "services/storageServices",
-        "",
+        "owner",
-        false,
         false,
+        true,
         false,
         false);
     this.supportsPatch = false;
@@ -71,15 +71,16 @@ public class StorageServiceResourceTest extends EntityResourceTest<StorageServic
   }
 
   @Test
-  void put_update_as_non_admin_401(TestInfo test) throws IOException {
+  void put_update_as_non_owner_401(TestInfo test) throws IOException {
-    createAndCheckEntity(createRequest(test).withDescription(null), ADMIN_AUTH_HEADERS);
+    createAndCheckEntity(createRequest(test).withDescription(null).withOwner(USER_OWNER1), ADMIN_AUTH_HEADERS);
 
-    // Update storage description and ingestion service that are null
+    // Update storage description and ingestion service as non owner should be forbidden
     HttpResponseException exception =
         assertThrows(
             HttpResponseException.class,
             () -> updateAndCheckEntity(createRequest(test), OK, TEST_AUTH_HEADERS, UpdateType.NO_CHANGE, null));
-    TestUtils.assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "is not admin");
+    TestUtils.assertResponse(
+        exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} " + "does not have permissions");
   }
 
   @Override
@@ -88,6 +89,7 @@ public class StorageServiceResourceTest extends EntityResourceTest<StorageServic
     return new CreateStorageService()
         .withName(name)
         .withServiceType(StorageServiceType.S3)
+        .withOwner(owner)
         .withDescription(description);
   }
 
@@ -95,7 +97,10 @@ public class StorageServiceResourceTest extends EntityResourceTest<StorageServic
   public void validateCreatedEntity(
       StorageService service, CreateStorageService createRequest, Map<String, String> authHeaders) {
     validateCommonEntityFields(
-        getEntityInterface(service), createRequest.getDescription(), getPrincipal(authHeaders), null);
+        getEntityInterface(service),
+        createRequest.getDescription(),
+        getPrincipal(authHeaders),
+        createRequest.getOwner());
     assertEquals(createRequest.getName(), service.getName());
   }
 
@@ -117,13 +122,14 @@ public class StorageServiceResourceTest extends EntityResourceTest<StorageServic
 
   @Override
   public void validateGetWithDifferentFields(StorageService service, boolean byName) throws HttpResponseException {
-    // No fields support
+    String fields = "owner";
     service =
         byName
-            ? getEntityByName(service.getName(), "", ADMIN_AUTH_HEADERS)
+            ? getEntityByName(service.getName(), fields, ADMIN_AUTH_HEADERS)
-            : getEntity(service.getId(), "", ADMIN_AUTH_HEADERS);
+            : getEntity(service.getId(), fields, ADMIN_AUTH_HEADERS);
     TestUtils.assertListNotNull(
         service.getHref(),
+        service.getOwner(),
         service.getVersion(),
         service.getUpdatedBy(),
         service.getServiceType(),