diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/Entity.java b/openmetadata-service/src/main/java/org/openmetadata/service/Entity.java
index b891dfa6d85..3824f1cfe99 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/Entity.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/Entity.java
@@ -15,6 +15,7 @@ package org.openmetadata.service;
 
 import static org.openmetadata.common.utils.CommonUtil.listOf;
 import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
+import static org.openmetadata.service.util.EntityUtil.getFlattenedEntityField;
 
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import io.github.classgraph.ClassGraph;
@@ -41,6 +42,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.core.Jdbi;
 import org.openmetadata.schema.EntityInterface;
 import org.openmetadata.schema.EntityTimeSeriesInterface;
+import org.openmetadata.schema.FieldInterface;
 import org.openmetadata.schema.entity.services.ServiceType;
 import org.openmetadata.schema.type.EntityReference;
 import org.openmetadata.schema.type.Include;
@@ -61,6 +63,7 @@ import org.openmetadata.service.jdbi3.UsageRepository;
 import org.openmetadata.service.resources.feeds.MessageParser.EntityLink;
 import org.openmetadata.service.search.SearchRepository;
 import org.openmetadata.service.util.EntityUtil.Fields;
+import org.openmetadata.service.util.FullyQualifiedName;
 
 @Slf4j
 public final class Entity {
@@ -529,4 +532,22 @@ public final class Entity {
       return classList.loadClasses();
     }
   }
+
+  public static <T extends FieldInterface> void populateEntityFieldTags(
+      String entityType, List<T> fields, String fqnPrefix, boolean setTags) {
+    EntityRepository<?> repository = Entity.getEntityRepository(entityType);
+    // Get Flattened Fields
+    List<T> flattenedFields = getFlattenedEntityField(fields);
+
+    // Fetch All tags belonging to Prefix
+    Map<String, List<TagLabel>> allTags = repository.getTagsByPrefix(fqnPrefix);
+    for (T c : listOrEmpty(flattenedFields)) {
+      if (setTags) {
+        List<TagLabel> columnTag = allTags.get(FullyQualifiedName.buildHash(c.getFullyQualifiedName()));
+        c.setTags(columnTag == null ? new ArrayList<>() : columnTag);
+      } else {
+        c.setTags(c.getTags());
+      }
+    }
+  }
 }
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/test/NoOpTestApplication.java b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/test/NoOpTestApplication.java
new file mode 100644
index 00000000000..94a0c0e8a7a
--- /dev/null
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/test/NoOpTestApplication.java
@@ -0,0 +1,17 @@
+package org.openmetadata.service.apps.bundles.test;
+
+import lombok.extern.slf4j.Slf4j;
+import org.openmetadata.schema.entity.app.App;
+import org.openmetadata.service.apps.AbstractNativeApplication;
+import org.openmetadata.service.jdbi3.CollectionDAO;
+import org.openmetadata.service.search.SearchRepository;
+
+@Slf4j
+public class NoOpTestApplication extends AbstractNativeApplication {
+
+  @Override
+  public void init(App app, CollectionDAO dao, SearchRepository searchRepository) {
+    super.init(app, dao, searchRepository);
+    LOG.info("NoOpTestApplication is initialized");
+  }
+}
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppMarketPlaceRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppMarketPlaceRepository.java
index 92b6ac14548..e5e197d5b6a 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppMarketPlaceRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppMarketPlaceRepository.java
@@ -16,6 +16,7 @@ public class AppMarketPlaceRepository extends EntityRepository<AppMarketPlaceDef
         "",
         "");
     supportsSearch = false;
+    quoteFqn = true;
   }
 
   @Override
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppRepository.java
index c44affb9917..94b6b0fe6c7 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/AppRepository.java
@@ -6,7 +6,6 @@ import static org.openmetadata.service.resources.teams.UserResource.getUser;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
-import javax.ws.rs.InternalServerErrorException;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.api.teams.CreateUser;
 import org.openmetadata.schema.auth.JWTAuthMechanism;
@@ -21,14 +20,12 @@ import org.openmetadata.schema.type.Include;
 import org.openmetadata.schema.type.ProviderType;
 import org.openmetadata.schema.type.Relationship;
 import org.openmetadata.service.Entity;
-import org.openmetadata.service.apps.scheduler.AppScheduler;
 import org.openmetadata.service.exception.EntityNotFoundException;
 import org.openmetadata.service.resources.apps.AppResource;
 import org.openmetadata.service.security.jwt.JWTTokenGenerator;
 import org.openmetadata.service.util.EntityUtil;
 import org.openmetadata.service.util.JsonUtils;
 import org.openmetadata.service.util.ResultList;
-import org.quartz.SchedulerException;
 
 @Slf4j
 public class AppRepository extends EntityRepository<App> {
@@ -45,6 +42,7 @@ public class AppRepository extends EntityRepository<App> {
         UPDATE_FIELDS,
         UPDATE_FIELDS);
     supportsSearch = false;
+    quoteFqn = true;
   }
 
   @Override
@@ -153,16 +151,6 @@ public class AppRepository extends EntityRepository<App> {
     entity.withBot(botUserRef).withOwner(ownerRef);
   }
 
-  @Override
-  public void postDelete(App entity) {
-    try {
-      AppScheduler.getInstance().deleteScheduledApplication(entity);
-    } catch (SchedulerException ex) {
-      LOG.error("Failed in delete Application from Scheduler.", ex);
-      throw new InternalServerErrorException("Failed in Delete App from Scheduler.");
-    }
-  }
-
   public EntityReference getBotUser(App application) {
     return application.getBot() != null
         ? application.getBot()
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java
index 93540494760..aad5bd5c100 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java
@@ -24,9 +24,11 @@ import static org.openmetadata.service.jdbi3.locator.ConnectionType.POSTGRES;
 import com.fasterxml.jackson.core.type.TypeReference;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -2201,10 +2203,66 @@ public interface CollectionDAO {
       return tags;
     }
 
+    default Map<String, List<TagLabel>> getTagsByPrefix(String targetFQNPrefix) {
+      Map<String, List<TagLabel>> resultSet = new LinkedHashMap<>();
+      List<Pair<String, TagLabel>> tags = getTagsInternalByPrefix(targetFQNPrefix);
+      tags.forEach(
+          pair -> {
+            String targetHash = pair.getLeft();
+            TagLabel tagLabel = pair.getRight();
+            List<TagLabel> listOfTarget = new ArrayList<>();
+            if (resultSet.containsKey(targetHash)) {
+              listOfTarget = resultSet.get(targetHash);
+              listOfTarget.add(tagLabel);
+            } else {
+              listOfTarget.add(tagLabel);
+            }
+            resultSet.put(targetHash, listOfTarget);
+          });
+      return resultSet;
+    }
+
     @SqlQuery(
         "SELECT source, tagFQN,  labelType, state FROM tag_usage WHERE targetFQNHash = :targetFQNHash ORDER BY tagFQN")
     List<TagLabel> getTagsInternal(@BindFQN("targetFQNHash") String targetFQNHash);
 
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT source, tagFQN, labelType, targetFQNHash, state, json "
+                + "FROM ("
+                + "  SELECT gterm.* , tu.* "
+                + "  FROM glossary_term_entity AS gterm "
+                + "  JOIN tag_usage AS tu "
+                + "  ON gterm.fqnHash = tu.tagFQNHash "
+                + "  WHERE tu.source = 1 "
+                + "  UNION ALL "
+                + "  SELECT ta.*, tu.* "
+                + "  FROM tag AS ta "
+                + "  JOIN tag_usage AS tu "
+                + "  ON ta.fqnHash = tu.tagFQNHash "
+                + "  WHERE tu.source = 0 "
+                + ") AS combined_data "
+                + "WHERE combined_data.targetFQNHash  LIKE CONCAT(:targetFQNHashPrefix, '.%')",
+        connectionType = MYSQL)
+    @ConnectionAwareSqlQuery(
+        value =
+            "SELECT source, tagFQN, labelType, targetFQNHash, state, json "
+                + "FROM ("
+                + "  SELECT gterm.*, tu.* "
+                + "  FROM glossary_term_entity AS gterm "
+                + "  JOIN tag_usage AS tu ON gterm.fqnHash = tu.tagFQNHash "
+                + "  WHERE tu.source = 1 "
+                + "  UNION ALL "
+                + "  SELECT ta.*, tu.* "
+                + "  FROM tag AS ta "
+                + "  JOIN tag_usage AS tu ON ta.fqnHash = tu.tagFQNHash "
+                + "  WHERE tu.source = 0 "
+                + ") AS combined_data "
+                + "WHERE combined_data.targetFQNHash LIKE CONCAT(:targetFQNHashPrefix, '.%')",
+        connectionType = POSTGRES)
+    @RegisterRowMapper(TagLabelRowMapperWithTargetFqnHash.class)
+    List<Pair<String, TagLabel>> getTagsInternalByPrefix(@BindFQN("targetFQNHashPrefix") String targetFQNHashPrefix);
+
     @SqlQuery("SELECT * FROM tag_usage")
     @Deprecated(since = "Release 1.1")
     @RegisterRowMapper(TagLabelMapperMigration.class)
@@ -2294,6 +2352,35 @@ public interface CollectionDAO {
       }
     }
 
+    class TagLabelRowMapperWithTargetFqnHash implements RowMapper<Pair<String, TagLabel>> {
+      @Override
+      public Pair<String, TagLabel> map(ResultSet r, StatementContext ctx) throws SQLException {
+        TagLabel label =
+            new TagLabel()
+                .withSource(TagLabel.TagSource.values()[r.getInt("source")])
+                .withLabelType(TagLabel.LabelType.values()[r.getInt("labelType")])
+                .withState(TagLabel.State.values()[r.getInt("state")])
+                .withTagFQN(r.getString("tagFQN"));
+        TagLabel.TagSource source = TagLabel.TagSource.values()[r.getInt("source")];
+        if (source == TagLabel.TagSource.CLASSIFICATION) {
+          Tag tag = JsonUtils.readValue(r.getString("json"), Tag.class);
+          label.setName(tag.getName());
+          label.setDisplayName(tag.getDisplayName());
+          label.setDescription(tag.getDescription());
+          label.setStyle(tag.getStyle());
+        } else if (source == TagLabel.TagSource.GLOSSARY) {
+          GlossaryTerm glossaryTerm = JsonUtils.readValue(r.getString("json"), GlossaryTerm.class);
+          label.setName(glossaryTerm.getName());
+          label.setDisplayName(glossaryTerm.getDisplayName());
+          label.setDescription(glossaryTerm.getDescription());
+          label.setStyle(glossaryTerm.getStyle());
+        } else {
+          throw new IllegalArgumentException("Invalid source type " + source);
+        }
+        return Pair.of(r.getString("targetFQNHash"), label);
+      }
+    }
+
     @Getter
     @Setter
     @Deprecated(since = "Release 1.1")
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ContainerRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ContainerRepository.java
index 20f02e37a6f..ae41596f7b0 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ContainerRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/ContainerRepository.java
@@ -8,6 +8,7 @@ import static org.openmetadata.service.Entity.DASHBOARD_DATA_MODEL;
 import static org.openmetadata.service.Entity.FIELD_PARENT;
 import static org.openmetadata.service.Entity.FIELD_TAGS;
 import static org.openmetadata.service.Entity.STORAGE_SERVICE;
+import static org.openmetadata.service.Entity.populateEntityFieldTags;
 
 import com.google.common.collect.Lists;
 import java.util.ArrayList;
@@ -54,7 +55,8 @@ public class ContainerRepository extends EntityRepository<Container> {
     setDefaultFields(container);
     container.setParent(fields.contains(FIELD_PARENT) ? getParent(container) : container.getParent());
     if (container.getDataModel() != null) {
-      populateDataModelColumnTags(fields.contains(FIELD_TAGS), container.getDataModel().getColumns());
+      populateDataModelColumnTags(
+          fields.contains(FIELD_TAGS), container.getFullyQualifiedName(), container.getDataModel().getColumns());
     }
     return container;
   }
@@ -65,11 +67,8 @@ public class ContainerRepository extends EntityRepository<Container> {
     return container.withDataModel(fields.contains("dataModel") ? container.getDataModel() : null);
   }
 
-  private void populateDataModelColumnTags(boolean setTags, List<Column> columns) {
-    for (Column c : listOrEmpty(columns)) {
-      c.setTags(setTags ? getTags(c.getFullyQualifiedName()) : null);
-      populateDataModelColumnTags(setTags, c.getChildren());
-    }
+  private void populateDataModelColumnTags(boolean setTags, String fqnPrefix, List<Column> columns) {
+    populateEntityFieldTags(entityType, columns, fqnPrefix, setTags);
   }
 
   private void setDefaultFields(Container container) {
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DashboardDataModelRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DashboardDataModelRepository.java
index d7d8c5fbf5a..ce66cb2b7db 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DashboardDataModelRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DashboardDataModelRepository.java
@@ -13,10 +13,10 @@
 
 package org.openmetadata.service.jdbi3;
 
-import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
 import static org.openmetadata.schema.type.Include.ALL;
 import static org.openmetadata.service.Entity.DASHBOARD_DATA_MODEL;
 import static org.openmetadata.service.Entity.FIELD_TAGS;
+import static org.openmetadata.service.Entity.populateEntityFieldTags;
 
 import java.util.List;
 import lombok.SneakyThrows;
@@ -161,7 +161,11 @@ public class DashboardDataModelRepository extends EntityRepository<DashboardData
 
   @Override
   public DashboardDataModel setFields(DashboardDataModel dashboardDataModel, Fields fields) {
-    getColumnTags(fields.contains(FIELD_TAGS), dashboardDataModel.getColumns());
+    populateEntityFieldTags(
+        entityType,
+        dashboardDataModel.getColumns(),
+        dashboardDataModel.getFullyQualifiedName(),
+        fields.contains(FIELD_TAGS));
     if (dashboardDataModel.getService() == null) {
       dashboardDataModel.withService(getContainer(dashboardDataModel.getId()));
     }
@@ -183,14 +187,6 @@ public class DashboardDataModelRepository extends EntityRepository<DashboardData
         .withId(original.getId());
   }
 
-  // TODO move this to base class?
-  private void getColumnTags(boolean setTags, List<Column> columns) {
-    for (Column c : listOrEmpty(columns)) {
-      c.setTags(setTags ? getTags(c.getFullyQualifiedName()) : c.getTags());
-      getColumnTags(setTags, c.getChildren());
-    }
-  }
-
   private void applyTags(List<Column> columns) {
     // Add column level tags by adding tag to column relationship
     for (Column column : columns) {
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java
index 4d12fde10a5..2fb3a9e3a36 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java
@@ -1296,6 +1296,10 @@ public abstract class EntityRepository<T extends EntityInterface> {
     return !supportsTags ? null : daoCollection.tagUsageDAO().getTags(fqn);
   }
 
+  public Map<String, List<TagLabel>> getTagsByPrefix(String prefix) {
+    return !supportsTags ? null : daoCollection.tagUsageDAO().getTagsByPrefix(prefix);
+  }
+
   protected List<EntityReference> getFollowers(T entity) {
     return !supportsFollower || entity == null
         ? Collections.emptyList()
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java
index da5b4c11faa..d166c42df60 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java
@@ -22,6 +22,7 @@ import static org.openmetadata.service.Entity.FIELD_OWNER;
 import static org.openmetadata.service.Entity.FIELD_TAGS;
 import static org.openmetadata.service.Entity.TABLE;
 import static org.openmetadata.service.Entity.getEntity;
+import static org.openmetadata.service.Entity.populateEntityFieldTags;
 import static org.openmetadata.service.util.LambdaExceptionUtil.ignoringComparator;
 import static org.openmetadata.service.util.LambdaExceptionUtil.rethrowFunction;
 
@@ -128,7 +129,8 @@ public class TableRepository extends EntityRepository<Table> {
     }
     if (fields.contains(COLUMN_FIELD)) {
       // We'll get column tags only if we are getting the column fields
-      getColumnTags(fields.contains(FIELD_TAGS), table.getColumns());
+      populateEntityFieldTags(
+          entityType, table.getColumns(), table.getFullyQualifiedName(), fields.contains(FIELD_TAGS));
     }
     table.setJoins(fields.contains("joins") ? getJoins(table) : table.getJoins());
     table.setTableProfilerConfig(
@@ -253,7 +255,7 @@ public class TableRepository extends EntityRepository<Table> {
 
     // Set the column tags. Will be used to mask the sample data
     if (!authorizePII) {
-      getColumnTags(true, table.getColumns());
+      populateEntityFieldTags(entityType, table.getColumns(), table.getFullyQualifiedName(), true);
       table.setTags(getTags(table));
       return PIIMasker.getSampleData(table);
     }
@@ -486,7 +488,7 @@ public class TableRepository extends EntityRepository<Table> {
 
     // Set the column tags. Will be used to hide the data
     if (!authorizePII) {
-      getColumnTags(true, table.getColumns());
+      populateEntityFieldTags(entityType, table.getColumns(), table.getFullyQualifiedName(), true);
       return PIIMasker.getTableProfile(table);
     }
 
@@ -803,14 +805,6 @@ public class TableRepository extends EntityRepository<Table> {
     return childrenColumn;
   }
 
-  // TODO duplicated code
-  private void getColumnTags(boolean setTags, List<Column> columns) {
-    for (Column c : listOrEmpty(columns)) {
-      c.setTags(setTags ? getTags(c.getFullyQualifiedName()) : c.getTags());
-      getColumnTags(setTags, c.getChildren());
-    }
-  }
-
   private void validateTableFQN(String fqn) {
     try {
       dao.existsByName(fqn);
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TopicRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TopicRepository.java
index 2c3a248e46e..e0296f5c10c 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TopicRepository.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TopicRepository.java
@@ -20,6 +20,7 @@ import static org.openmetadata.service.Entity.FIELD_DESCRIPTION;
 import static org.openmetadata.service.Entity.FIELD_DISPLAY_NAME;
 import static org.openmetadata.service.Entity.FIELD_TAGS;
 import static org.openmetadata.service.Entity.MESSAGING_SERVICE;
+import static org.openmetadata.service.Entity.populateEntityFieldTags;
 
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -121,7 +122,11 @@ public class TopicRepository extends EntityRepository<Topic> {
   public Topic setFields(Topic topic, Fields fields) {
     topic.setService(getContainer(topic.getId()));
     if (topic.getMessageSchema() != null) {
-      getFieldTags(fields.contains(FIELD_TAGS), topic.getMessageSchema().getSchemaFields());
+      populateEntityFieldTags(
+          entityType,
+          topic.getMessageSchema().getSchemaFields(),
+          topic.getFullyQualifiedName(),
+          fields.contains(FIELD_TAGS));
     }
     return topic;
   }
@@ -155,7 +160,8 @@ public class TopicRepository extends EntityRepository<Topic> {
 
     // Set the fields tags. Will be used to mask the sample data
     if (!authorizePII) {
-      getFieldTags(true, topic.getMessageSchema().getSchemaFields());
+      populateEntityFieldTags(
+          entityType, topic.getMessageSchema().getSchemaFields(), topic.getFullyQualifiedName(), true);
       topic.setTags(getTags(topic));
       return PIIMasker.getSampleData(topic);
     }
@@ -185,15 +191,6 @@ public class TopicRepository extends EntityRepository<Topic> {
         });
   }
 
-  private void getFieldTags(boolean setTags, List<Field> fields) {
-    for (Field f : listOrEmpty(fields)) {
-      if (f.getTags() == null) {
-        f.setTags(setTags ? getTags(f.getFullyQualifiedName()) : null);
-        getFieldTags(setTags, f.getChildren());
-      }
-    }
-  }
-
   private void addDerivedFieldTags(List<Field> fields) {
     if (nullOrEmpty(fields)) {
       return;
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppMarketPlaceResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppMarketPlaceResource.java
index e2b6191c6ba..b5b5b66b73d 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppMarketPlaceResource.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppMarketPlaceResource.java
@@ -36,6 +36,7 @@ import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
 import javax.ws.rs.core.UriInfo;
 import lombok.extern.slf4j.Slf4j;
+import org.openmetadata.schema.api.data.RestoreEntity;
 import org.openmetadata.schema.entity.app.App;
 import org.openmetadata.schema.entity.app.AppMarketPlaceDefinition;
 import org.openmetadata.schema.entity.app.AppType;
@@ -59,12 +60,13 @@ import org.openmetadata.service.util.ResultList;
 @Tag(name = "Apps", description = "Apps marketplace holds to application available for Open-metadata")
 @Produces(MediaType.APPLICATION_JSON)
 @Consumes(MediaType.APPLICATION_JSON)
-@Collection(name = "appsMarketPlace", order = 8)
+@Collection(name = "apps/marketplace", order = 8)
 @Slf4j
 public class AppMarketPlaceResource extends EntityResource<AppMarketPlaceDefinition, AppMarketPlaceRepository> {
   public static final String COLLECTION_PATH = "/v1/apps/marketplace/";
   private PipelineServiceClient pipelineServiceClient;
-  static final String FIELDS = "owner";
+
+  static final String FIELDS = "owner,tags";
 
   @Override
   public void initialize(OpenMetadataApplicationConfig config) {
@@ -365,6 +367,26 @@ public class AppMarketPlaceResource extends EntityResource<AppMarketPlaceDefinit
     return delete(uriInfo, securityContext, id, true, hardDelete);
   }
 
+  @PUT
+  @Path("/restore")
+  @Operation(
+      operationId = "restore",
+      summary = "Restore a soft deleted KPI",
+      description = "Restore a soft deleted App.",
+      responses = {
+        @ApiResponse(
+            responseCode = "200",
+            description = "Successfully restored the App. ",
+            content =
+                @Content(
+                    mediaType = "application/json",
+                    schema = @Schema(implementation = AppMarketPlaceDefinition.class)))
+      })
+  public Response restoreApp(
+      @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid RestoreEntity restore) {
+    return restoreEntity(uriInfo, securityContext, restore.getId());
+  }
+
   private AppMarketPlaceDefinition getApplicationDefinition(
       CreateAppMarketPlaceDefinitionReq create, String updatedBy) {
     AppMarketPlaceDefinition app =
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppResource.java b/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppResource.java
index ac0de81bedb..c6be255d8cf 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppResource.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppResource.java
@@ -1,5 +1,6 @@
 package org.openmetadata.service.resources.apps;
 
+import static org.openmetadata.common.utils.CommonUtil.nullOrEmpty;
 import static org.openmetadata.schema.type.Include.ALL;
 import static org.openmetadata.service.Entity.APPLICATION;
 import static org.openmetadata.service.Entity.BOT;
@@ -28,6 +29,7 @@ import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
+import javax.ws.rs.InternalServerErrorException;
 import javax.ws.rs.PATCH;
 import javax.ws.rs.POST;
 import javax.ws.rs.PUT;
@@ -458,7 +460,6 @@ public class AppResource extends EntityResource<App, AppRepository> {
   }
 
   @POST
-  @Path("/install")
   @Operation(
       operationId = "createApplication",
       summary = "Create a Application",
@@ -552,7 +553,11 @@ public class AppResource extends EntityResource<App, AppRepository> {
           @DefaultValue("false")
           boolean hardDelete,
       @Parameter(description = "Name of the App", schema = @Schema(type = "string")) @PathParam("name") String name) {
-    return deleteByName(uriInfo, securityContext, name, true, hardDelete);
+    Response response = deleteByName(uriInfo, securityContext, name, true, hardDelete);
+    if (response.getStatus() == Response.Status.OK.getStatusCode()) {
+      deleteApp(securityContext, (App) response.getEntity(), hardDelete);
+    }
+    return response;
   }
 
   @DELETE
@@ -573,7 +578,11 @@ public class AppResource extends EntityResource<App, AppRepository> {
           @DefaultValue("false")
           boolean hardDelete,
       @Parameter(description = "Id of the App", schema = @Schema(type = "UUID")) @PathParam("id") UUID id) {
-    return delete(uriInfo, securityContext, id, true, hardDelete);
+    Response response = delete(uriInfo, securityContext, id, true, hardDelete);
+    if (response.getStatus() == Response.Status.OK.getStatusCode()) {
+      deleteApp(securityContext, (App) response.getEntity(), hardDelete);
+    }
+    return response;
   }
 
   @PUT
@@ -590,7 +599,14 @@ public class AppResource extends EntityResource<App, AppRepository> {
       })
   public Response restoreApp(
       @Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid RestoreEntity restore) {
-    return restoreEntity(uriInfo, securityContext, restore.getId());
+    Response response = restoreEntity(uriInfo, securityContext, restore.getId());
+    if (response.getStatus() == Response.Status.OK.getStatusCode()) {
+      App app = (App) response.getEntity();
+      if (app.getScheduleType().equals(ScheduleType.Scheduled)) {
+        ApplicationHandler.scheduleApplication(app, Entity.getCollectionDAO(), searchRepository);
+      }
+    }
+    return response;
   }
 
   @POST
@@ -734,8 +750,8 @@ public class AppResource extends EntityResource<App, AppRepository> {
         new App()
             .withId(UUID.randomUUID())
             .withName(marketPlaceDefinition.getName())
-            .withDisplayName(marketPlaceDefinition.getDisplayName())
-            .withDescription(marketPlaceDefinition.getDescription())
+            .withDisplayName(createAppRequest.getDisplayName())
+            .withDescription(createAppRequest.getDescription())
             .withOwner(owner)
             .withUpdatedBy(updatedBy)
             .withUpdatedAt(System.currentTimeMillis())
@@ -767,4 +783,37 @@ public class AppResource extends EntityResource<App, AppRepository> {
       app.setBot(repository.createNewAppBot(app));
     }
   }
+
+  private void deleteApp(SecurityContext securityContext, App installedApp, boolean hardDelete) {
+    if (installedApp.getAppType().equals(AppType.Internal)) {
+      try {
+        AppScheduler.getInstance().deleteScheduledApplication(installedApp);
+      } catch (SchedulerException ex) {
+        LOG.error("Failed in delete Application from Scheduler.", ex);
+        throw new InternalServerErrorException("Failed in Delete App from Scheduler.");
+      }
+    } else {
+      App app = repository.getByName(null, installedApp.getName(), repository.getFields("bot,pipelines"));
+      if (!nullOrEmpty(app.getPipelines())) {
+        EntityReference pipelineRef = app.getPipelines().get(0);
+        IngestionPipelineRepository ingestionPipelineRepository =
+            (IngestionPipelineRepository) Entity.getEntityRepository(Entity.INGESTION_PIPELINE);
+
+        IngestionPipeline ingestionPipeline =
+            ingestionPipelineRepository.get(
+                null, pipelineRef.getId(), ingestionPipelineRepository.getFields(FIELD_OWNER));
+
+        if (hardDelete) {
+          // Remove the Pipeline in case of Delete
+          if (!nullOrEmpty(app.getPipelines())) {
+            pipelineServiceClient.deletePipeline(ingestionPipeline);
+          }
+        } else {
+          // Just Kill Running ingestion
+          decryptOrNullify(securityContext, ingestionPipeline, app.getBot().getName(), true);
+          pipelineServiceClient.killIngestion(ingestionPipeline);
+        }
+      }
+    }
+  }
 }
diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/util/EntityUtil.java b/openmetadata-service/src/main/java/org/openmetadata/service/util/EntityUtil.java
index e523142db87..0ff7f1db0bd 100644
--- a/openmetadata-service/src/main/java/org/openmetadata/service/util/EntityUtil.java
+++ b/openmetadata-service/src/main/java/org/openmetadata/service/util/EntityUtil.java
@@ -13,6 +13,7 @@
 
 package org.openmetadata.service.util;
 
+import static org.openmetadata.common.utils.CommonUtil.listOrEmpty;
 import static org.openmetadata.common.utils.CommonUtil.nullOrEmpty;
 import static org.openmetadata.schema.type.Include.ALL;
 
@@ -38,6 +39,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.codec.binary.Hex;
 import org.openmetadata.common.utils.CommonUtil;
 import org.openmetadata.schema.EntityInterface;
+import org.openmetadata.schema.FieldInterface;
 import org.openmetadata.schema.api.data.TermReference;
 import org.openmetadata.schema.entity.classification.Tag;
 import org.openmetadata.schema.entity.data.GlossaryTerm;
@@ -556,4 +558,18 @@ public final class EntityUtil {
         .orElseThrow(
             () -> new IllegalArgumentException(CatalogExceptionMessage.invalidFieldName("column", columnName)));
   }
+
+  public static <T extends FieldInterface> List<T> getFlattenedEntityField(List<T> fields) {
+    List<T> flattenedFields = new ArrayList<>();
+    fields.forEach(column -> flattenEntityField(column, flattenedFields));
+    return flattenedFields;
+  }
+
+  private static <T extends FieldInterface> void flattenEntityField(T field, List<T> flattenedFields) {
+    flattenedFields.add(field);
+    List<T> children = (List<T>) field.getChildren();
+    for (T child : listOrEmpty(children)) {
+      flattenEntityField(child, flattenedFields);
+    }
+  }
 }
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java
index c383f002352..098bef48c72 100644
--- a/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java
+++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/EntityResourceTest.java
@@ -428,21 +428,21 @@ public abstract class EntityResourceTest<T extends EntityInterface, K extends Cr
     new QueryResourceTest().setupQuery(test);
 
     runWebhookTests = new Random().nextBoolean();
-    if (runWebhookTests) {
-      webhookCallbackResource.clearEvents();
-      EventSubscriptionResourceTest alertResourceTest = new EventSubscriptionResourceTest();
-      alertResourceTest.startWebhookSubscription();
-      alertResourceTest.startWebhookEntitySubscriptions(entityType);
-    }
+    // if (true) {
+    webhookCallbackResource.clearEvents();
+    EventSubscriptionResourceTest alertResourceTest = new EventSubscriptionResourceTest();
+    alertResourceTest.startWebhookSubscription();
+    alertResourceTest.startWebhookEntitySubscriptions(entityType);
+    // }
   }
 
   @AfterAll
   public void afterAllTests() throws Exception {
-    if (runWebhookTests) {
-      EventSubscriptionResourceTest alertResourceTest = new EventSubscriptionResourceTest();
-      alertResourceTest.validateWebhookEvents();
-      alertResourceTest.validateWebhookEntityEvents(entityType);
-    }
+    // if (true) {
+    EventSubscriptionResourceTest alertResourceTest = new EventSubscriptionResourceTest();
+    alertResourceTest.validateWebhookEvents();
+    alertResourceTest.validateWebhookEntityEvents(entityType);
+    // }
     delete_recursiveTest();
   }
 
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/apps/AppMarketPlaceResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/apps/AppMarketPlaceResourceTest.java
new file mode 100644
index 00000000000..2239f80a594
--- /dev/null
+++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/apps/AppMarketPlaceResourceTest.java
@@ -0,0 +1,113 @@
+package org.openmetadata.service.resources.apps;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.openmetadata.service.util.TestUtils.ADMIN_AUTH_HEADERS;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.client.HttpResponseException;
+import org.openmetadata.schema.entity.app.AppConfiguration;
+import org.openmetadata.schema.entity.app.AppMarketPlaceDefinition;
+import org.openmetadata.schema.entity.app.AppType;
+import org.openmetadata.schema.entity.app.CreateAppMarketPlaceDefinitionReq;
+import org.openmetadata.schema.entity.app.NativeAppPermission;
+import org.openmetadata.schema.entity.app.ScheduleType;
+import org.openmetadata.schema.entity.app.ScheduledExecutionContext;
+import org.openmetadata.service.Entity;
+import org.openmetadata.service.resources.EntityResourceTest;
+import org.openmetadata.service.util.TestUtils;
+
+@Slf4j
+public class AppMarketPlaceResourceTest
+    extends EntityResourceTest<AppMarketPlaceDefinition, CreateAppMarketPlaceDefinitionReq> {
+
+  public AppMarketPlaceResourceTest() {
+    super(
+        Entity.APP_MARKET_PLACE_DEF,
+        AppMarketPlaceDefinition.class,
+        AppMarketPlaceResource.AppMarketPlaceDefinitionList.class,
+        "apps/marketplace",
+        AppMarketPlaceResource.FIELDS);
+    supportsFieldsQueryParam = false;
+    supportedNameCharacters = "_-.";
+  }
+
+  @Override
+  public CreateAppMarketPlaceDefinitionReq createRequest(String name) {
+    try {
+      return new CreateAppMarketPlaceDefinitionReq()
+          .withName(name)
+          .withOwner(USER1_REF)
+          .withDeveloper("OM")
+          .withDeveloperUrl("https://test.com")
+          .withSupportEmail("test@openmetadata.org")
+          .withPrivacyPolicyUrl("https://privacy@openmetadata.org")
+          .withClassName("org.openmetadata.service.apps.bundles.test.NoOpTestApplication")
+          .withAppType(AppType.Internal)
+          .withScheduleType(ScheduleType.Scheduled)
+          .withRuntime(new ScheduledExecutionContext().withEnabled(true))
+          .withAppConfiguration(new AppConfiguration().withAdditionalProperty("test", "20"))
+          .withPermission(NativeAppPermission.All)
+          .withAppLogoUrl(new URI("https://test.com"))
+          .withAppScreenshots(new HashSet<>(List.of("AppLogo")))
+          .withFeatures("App Features");
+    } catch (URISyntaxException ex) {
+      LOG.error("Encountered error in Create Request for AppMarketPlaceResourceTest for App Logo Url.");
+    }
+    return null;
+  }
+
+  @Override
+  public void validateCreatedEntity(
+      AppMarketPlaceDefinition createdEntity,
+      CreateAppMarketPlaceDefinitionReq request,
+      Map<String, String> authHeaders)
+      throws HttpResponseException {}
+
+  @Override
+  public void compareEntities(
+      AppMarketPlaceDefinition expected, AppMarketPlaceDefinition updated, Map<String, String> authHeaders)
+      throws HttpResponseException {
+    assertEquals(expected.getDeveloper(), updated.getDeveloper());
+    assertEquals(expected.getDeveloperUrl(), updated.getDeveloperUrl());
+    assertEquals(expected.getSupportEmail(), updated.getSupportEmail());
+    assertEquals(expected.getPrivacyPolicyUrl(), updated.getPrivacyPolicyUrl());
+    assertEquals(expected.getClassName(), updated.getClassName());
+    assertEquals(expected.getAppType(), updated.getAppType());
+    assertEquals(expected.getScheduleType(), updated.getScheduleType());
+    assertEquals(expected.getAppConfiguration(), updated.getAppConfiguration());
+    assertEquals(expected.getPermission(), updated.getPermission());
+    assertEquals(expected.getAppLogoUrl(), updated.getAppLogoUrl());
+    assertEquals(expected.getAppScreenshots(), updated.getAppScreenshots());
+    assertEquals(expected.getAppScreenshots(), updated.getAppScreenshots());
+    assertEquals(expected.getFeatures(), updated.getFeatures());
+  }
+
+  @Override
+  public AppMarketPlaceDefinition validateGetWithDifferentFields(AppMarketPlaceDefinition entity, boolean byName)
+      throws HttpResponseException {
+    String fields = "";
+    entity =
+        byName
+            ? getEntityByName(entity.getFullyQualifiedName(), fields, ADMIN_AUTH_HEADERS)
+            : getEntity(entity.getId(), fields, ADMIN_AUTH_HEADERS);
+    TestUtils.assertListNull(entity.getOwner());
+
+    fields = "owner,tags";
+    entity =
+        byName
+            ? getEntityByName(entity.getFullyQualifiedName(), fields, ADMIN_AUTH_HEADERS)
+            : getEntity(entity.getId(), fields, ADMIN_AUTH_HEADERS);
+    return entity;
+  }
+
+  @Override
+  public void assertFieldChange(String fieldName, Object expected, Object actual) throws IOException {
+    assertCommonFieldChange(fieldName, expected, actual);
+  }
+}
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/apps/AppsResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/apps/AppsResourceTest.java
new file mode 100644
index 00000000000..62ffedb3302
--- /dev/null
+++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/apps/AppsResourceTest.java
@@ -0,0 +1,83 @@
+package org.openmetadata.service.resources.apps;
+
+import static org.openmetadata.service.util.TestUtils.ADMIN_AUTH_HEADERS;
+
+import java.io.IOException;
+import java.util.Map;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.client.HttpResponseException;
+import org.junit.jupiter.api.Test;
+import org.openmetadata.schema.entity.app.App;
+import org.openmetadata.schema.entity.app.AppMarketPlaceDefinition;
+import org.openmetadata.schema.entity.app.AppSchedule;
+import org.openmetadata.schema.entity.app.CreateApp;
+import org.openmetadata.schema.entity.app.CreateAppMarketPlaceDefinitionReq;
+import org.openmetadata.service.Entity;
+import org.openmetadata.service.exception.EntityNotFoundException;
+import org.openmetadata.service.resources.EntityResourceTest;
+import org.openmetadata.service.util.TestUtils;
+
+@Slf4j
+public class AppsResourceTest extends EntityResourceTest<App, CreateApp> {
+
+  public AppsResourceTest() {
+    super(Entity.APPLICATION, App.class, AppResource.AppList.class, "apps", AppResource.FIELDS);
+    supportsFieldsQueryParam = false;
+    supportedNameCharacters = "_-.";
+  }
+
+  @Override
+  @SneakyThrows
+  public CreateApp createRequest(String name) {
+    // Create AppMarketPlaceDefinition
+    AppMarketPlaceResourceTest appMarketPlaceResourceTest = new AppMarketPlaceResourceTest();
+    AppMarketPlaceDefinition appMarketPlaceDefinition = null;
+    try {
+      appMarketPlaceDefinition = appMarketPlaceResourceTest.getEntityByName(name, ADMIN_AUTH_HEADERS);
+    } catch (EntityNotFoundException | HttpResponseException ex) {
+      CreateAppMarketPlaceDefinitionReq req = appMarketPlaceResourceTest.createRequest(name);
+      appMarketPlaceDefinition = appMarketPlaceResourceTest.createAndCheckEntity(req, ADMIN_AUTH_HEADERS);
+    }
+    // Create Request
+    return new CreateApp()
+        .withName(appMarketPlaceDefinition.getName())
+        .withAppConfiguration(appMarketPlaceDefinition.getAppConfiguration())
+        .withAppSchedule(new AppSchedule().withScheduleType(AppSchedule.ScheduleTimeline.HOURLY));
+  }
+
+  @Test
+  @SneakyThrows
+  @Override
+  protected void post_entityCreateWithInvalidName_400() {
+    // Does not apply since the App is already validated in the AppMarketDefinition
+  }
+
+  @Override
+  public void validateCreatedEntity(App createdEntity, CreateApp request, Map<String, String> authHeaders)
+      throws HttpResponseException {}
+
+  @Override
+  public void compareEntities(App expected, App updated, Map<String, String> authHeaders)
+      throws HttpResponseException {}
+
+  @Override
+  public App validateGetWithDifferentFields(App entity, boolean byName) throws HttpResponseException {
+    String fields = "";
+    entity =
+        byName
+            ? getEntityByName(entity.getFullyQualifiedName(), fields, ADMIN_AUTH_HEADERS)
+            : getEntity(entity.getId(), fields, ADMIN_AUTH_HEADERS);
+    TestUtils.assertListNull(entity.getOwner());
+
+    fields = "owner";
+    entity =
+        byName
+            ? getEntityByName(entity.getFullyQualifiedName(), fields, ADMIN_AUTH_HEADERS)
+            : getEntity(entity.getId(), fields, ADMIN_AUTH_HEADERS);
+    return entity;
+  }
+
+  @Override
+  public void assertFieldChange(String fieldName, Object expected, Object actual) throws IOException {}
+}
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/bots/BotResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/bots/BotResourceTest.java
index 676fbdbb219..2999e56feca 100644
--- a/openmetadata-service/src/test/java/org/openmetadata/service/resources/bots/BotResourceTest.java
+++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/bots/BotResourceTest.java
@@ -7,9 +7,13 @@ import static org.openmetadata.service.util.TestUtils.INGESTION_BOT;
 import static org.openmetadata.service.util.TestUtils.assertResponse;
 
 import java.io.IOException;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.http.client.HttpResponseException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -17,17 +21,20 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.openmetadata.schema.api.CreateBot;
 import org.openmetadata.schema.entity.Bot;
+import org.openmetadata.schema.entity.app.App;
 import org.openmetadata.schema.entity.teams.User;
 import org.openmetadata.schema.type.ProviderType;
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.exception.CatalogExceptionMessage;
 import org.openmetadata.service.jdbi3.EntityRepository;
 import org.openmetadata.service.resources.EntityResourceTest;
+import org.openmetadata.service.resources.apps.AppsResourceTest;
 import org.openmetadata.service.resources.bots.BotResource.BotList;
 import org.openmetadata.service.resources.teams.UserResourceTest;
 import org.openmetadata.service.util.ResultList;
 import org.openmetadata.service.util.TestUtils;
 
+@Slf4j
 public class BotResourceTest extends EntityResourceTest<Bot, CreateBot> {
   public static User botUser;
 
@@ -44,9 +51,24 @@ public class BotResourceTest extends EntityResourceTest<Bot, CreateBot> {
   @BeforeEach
   public void beforeEach() throws HttpResponseException {
     ResultList<Bot> bots = listEntities(null, ADMIN_AUTH_HEADERS);
+
+    // Get App Bots
+    AppsResourceTest appsResourceTest = new AppsResourceTest();
+    ResultList<App> appResultList = appsResourceTest.listEntities(null, ADMIN_AUTH_HEADERS);
+    Set<UUID> applicationBotIds = new HashSet<>();
+    appResultList
+        .getData()
+        .forEach(
+            app -> {
+              if (app.getBot() != null) {
+                applicationBotIds.add(app.getBot().getId());
+              } else {
+                LOG.error("Bot Entry Null for App : {}", app.getName());
+              }
+            });
     for (Bot bot : bots.getData()) {
       try {
-        if (!bot.getProvider().equals(ProviderType.SYSTEM)) {
+        if (!bot.getProvider().equals(ProviderType.SYSTEM) && !applicationBotIds.contains(bot.getId())) {
           deleteEntity(bot.getId(), true, true, ADMIN_AUTH_HEADERS);
           createUser();
         }
diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/resources/events/EventSubscriptionResourceTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/resources/events/EventSubscriptionResourceTest.java
index 39813d90af4..d5b8aec6364 100644
--- a/openmetadata-service/src/test/java/org/openmetadata/service/resources/events/EventSubscriptionResourceTest.java
+++ b/openmetadata-service/src/test/java/org/openmetadata/service/resources/events/EventSubscriptionResourceTest.java
@@ -407,6 +407,22 @@ public class EventSubscriptionResourceTest extends EntityResourceTest<EventSubsc
     List<ChangeEvent> expected =
         getChangeEvents(entityCreated, entityUpdated, entityRestored, entityDeleted, timestamp, ADMIN_AUTH_HEADERS)
             .getData();
+
+    // Comparison if all callBack Event are there in expected
+    for (ChangeEvent changeEvent : callbackEvents) {
+      boolean found = false;
+      for (ChangeEvent expectedChangeEvent : expected) {
+        if (changeEvent.getEventType().equals(expectedChangeEvent.getEventType())
+            && changeEvent.getEntityId().equals(expectedChangeEvent.getEntityId())) {
+          found = true;
+          break;
+        }
+      }
+      if (!found) {
+        LOG.error("[ChangeEventError] Change Events Missing from Expected: {}", changeEvent.toString());
+      }
+    }
+
     Awaitility.await()
         .pollInterval(Duration.ofMillis(100L))
         .atMost(Duration.ofMillis(iteration * 100L))
diff --git a/openmetadata-spec/src/main/java/org/openmetadata/schema/FieldInterface.java b/openmetadata-spec/src/main/java/org/openmetadata/schema/FieldInterface.java
new file mode 100644
index 00000000000..52576ed22f0
--- /dev/null
+++ b/openmetadata-spec/src/main/java/org/openmetadata/schema/FieldInterface.java
@@ -0,0 +1,24 @@
+package org.openmetadata.schema;
+
+import java.util.List;
+import org.openmetadata.schema.type.TagLabel;
+
+public interface FieldInterface {
+  String getName();
+
+  String getDisplayName();
+
+  String getDescription();
+
+  String getDataTypeDisplay();
+
+  String getFullyQualifiedName();
+
+  List<TagLabel> getTags();
+
+  default void setTags(List<TagLabel> tags) {
+    /* no-op implementation to be overridden */
+  }
+
+  List<? extends FieldInterface> getChildren();
+}
diff --git a/openmetadata-spec/src/main/resources/json/schema/entity/applications/createAppRequest.json b/openmetadata-spec/src/main/resources/json/schema/entity/applications/createAppRequest.json
index a605c4c7500..07a82aca6db 100644
--- a/openmetadata-spec/src/main/resources/json/schema/entity/applications/createAppRequest.json
+++ b/openmetadata-spec/src/main/resources/json/schema/entity/applications/createAppRequest.json
@@ -3,6 +3,7 @@
   "$schema": "http://json-schema.org/draft-07/schema#",
   "title": "CreateApp",
   "javaType": "org.openmetadata.schema.entity.app.CreateApp",
+  "javaInterfaces": ["org.openmetadata.schema.CreateEntity"],
   "description": "This schema defines the create applications request for Open-Metadata.",
   "type": "object",
   "properties": {
@@ -10,6 +11,14 @@
       "description": "Name of the Application.",
       "$ref": "../../type/basic.json#/definitions/entityName"
     },
+    "displayName": {
+      "description": "Display Name for the application.",
+      "type": "string"
+    },
+    "description": {
+      "description": "Description of the Application.",
+      "$ref": "../../type/basic.json#/definitions/markdown"
+    },
     "owner": {
       "description": "Owner of this workflow.",
       "$ref": "../../type/entityReference.json",
diff --git a/openmetadata-spec/src/main/resources/json/schema/entity/data/searchIndex.json b/openmetadata-spec/src/main/resources/json/schema/entity/data/searchIndex.json
index a155ccc4c03..f87ff88b2d5 100644
--- a/openmetadata-spec/src/main/resources/json/schema/entity/data/searchIndex.json
+++ b/openmetadata-spec/src/main/resources/json/schema/entity/data/searchIndex.json
@@ -80,6 +80,7 @@
     "searchIndexField": {
       "type": "object",
       "javaType": "org.openmetadata.schema.type.SearchIndexField",
+      "javaInterfaces": ["org.openmetadata.schema.FieldInterface"],
       "description": "This schema defines the type for a field in a searchIndex.",
       "properties": {
         "name": {
diff --git a/openmetadata-spec/src/main/resources/json/schema/entity/data/table.json b/openmetadata-spec/src/main/resources/json/schema/entity/data/table.json
index 8bcb19dc85b..6236552421b 100644
--- a/openmetadata-spec/src/main/resources/json/schema/entity/data/table.json
+++ b/openmetadata-spec/src/main/resources/json/schema/entity/data/table.json
@@ -241,6 +241,7 @@
     "column": {
       "type": "object",
       "javaType": "org.openmetadata.schema.type.Column",
+      "javaInterfaces": ["org.openmetadata.schema.FieldInterface"],
       "description": "This schema defines the type for a column in a table.",
       "properties": {
         "name": {
diff --git a/openmetadata-spec/src/main/resources/json/schema/type/schema.json b/openmetadata-spec/src/main/resources/json/schema/type/schema.json
index 29eedff7c41..526961c31cb 100644
--- a/openmetadata-spec/src/main/resources/json/schema/type/schema.json
+++ b/openmetadata-spec/src/main/resources/json/schema/type/schema.json
@@ -71,6 +71,7 @@
     "field": {
       "type": "object",
       "javaType": "org.openmetadata.schema.type.Field",
+      "javaInterfaces": ["org.openmetadata.schema.FieldInterface"],
       "description": "This schema defines the nested object to capture protobuf/avro/jsonschema of topic's schema.",
       "properties": {
         "name": {
diff --git a/openmetadata-ui/src/main/resources/ui/src/pages/AppInstall/AppInstall.component.tsx b/openmetadata-ui/src/main/resources/ui/src/pages/AppInstall/AppInstall.component.tsx
index f88fb03f44d..d6eccde370b 100644
--- a/openmetadata-ui/src/main/resources/ui/src/pages/AppInstall/AppInstall.component.tsx
+++ b/openmetadata-ui/src/main/resources/ui/src/pages/AppInstall/AppInstall.component.tsx
@@ -113,6 +113,8 @@ const AppInstall = () => {
           cronExpression: repeatFrequency,
         },
         name: fqn,
+        description: appData?.description,
+        displayName: appData?.displayName,
       };
       await installApplication(data);
 
diff --git a/openmetadata-ui/src/main/resources/ui/src/rest/applicationAPI.ts b/openmetadata-ui/src/main/resources/ui/src/rest/applicationAPI.ts
index afe45a0ff18..b6a335e9af3 100644
--- a/openmetadata-ui/src/main/resources/ui/src/rest/applicationAPI.ts
+++ b/openmetadata-ui/src/main/resources/ui/src/rest/applicationAPI.ts
@@ -40,7 +40,7 @@ export const getApplicationList = async (params?: ListParams) => {
 export const installApplication = (
   data: CreateAppRequest
 ): Promise<AxiosResponse> => {
-  return APIClient.post(`${BASE_URL}/install`, data);
+  return APIClient.post(`${BASE_URL}`, data);
 };
 
 export const getApplicationByName = async (