Fix #5764 Backend: Add support for filter by task status (#5899)

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

@@ -69,6 +69,7 @@ import org.openmetadata.catalog.jdbi3.locator.ConnectionAwareSqlUpdate;
 import org.openmetadata.catalog.type.Relationship;
 import org.openmetadata.catalog.type.TagCategory;
 import org.openmetadata.catalog.type.TagLabel;
+import org.openmetadata.catalog.type.TaskStatus;
 import org.openmetadata.catalog.type.ThreadType;
 import org.openmetadata.catalog.type.UsageDetails;
 import org.openmetadata.catalog.type.UsageStats;
@@ -517,8 +518,10 @@ public interface CollectionDAO {
     @SqlQuery("SELECT json FROM thread_entity ORDER BY updatedAt DESC")
     List<String> list();
 
-    @SqlQuery("SELECT count(id) FROM thread_entity WHERE resolved = :resolved AND (:type IS NULL OR type = :type)")
+    @SqlQuery(
-    int listCount(@Bind("resolved") boolean resolved, @Bind("type") ThreadType type);
+        "SELECT count(id) FROM thread_entity WHERE resolved = :resolved "
+            + "AND (:type IS NULL OR type = :type) AND (:status IS NULL OR taskStatus = :status)")
+    int listCount(@Bind("status") TaskStatus status, @Bind("resolved") boolean resolved, @Bind("type") ThreadType type);
 
     @ConnectionAwareSqlUpdate(value = "UPDATE task_sequence SET id=LAST_INSERT_ID(id+1)", connectionType = MYSQL)
     @ConnectionAwareSqlUpdate(value = "UPDATE task_sequence SET id=(id+1) RETURNING id", connectionType = POSTGRES)
@@ -532,36 +535,40 @@ public interface CollectionDAO {
 
     @SqlQuery(
         "SELECT json FROM thread_entity "
-            + "WHERE updatedAt > :before AND resolved = :resolved AND (:type IS NULL OR type = :type) "
+            + "WHERE updatedAt > :before AND resolved = :resolved "
+            + "AND (:type IS NULL OR type = :type) AND (:status IS NULL OR taskStatus = :status) "
             + "ORDER BY updatedAt DESC "
             + "LIMIT :limit")
     List<String> listBefore(
         @Bind("limit") int limit,
         @Bind("before") long before,
+        @Bind("status") TaskStatus status,
         @Bind("resolved") boolean resolved,
         @Bind("type") ThreadType type);
 
     @SqlQuery(
         "SELECT json FROM thread_entity "
-            + "WHERE updatedAt < :after AND resolved = :resolved AND (:type IS NULL OR type = :type) "
+            + "WHERE updatedAt < :after AND resolved = :resolved "
+            + "AND (:type IS NULL OR type = :type) AND (:status IS NULL OR taskStatus = :status) "
             + "ORDER BY updatedAt DESC "
             + "LIMIT :limit")
     List<String> listAfter(
         @Bind("limit") int limit,
         @Bind("after") long after,
+        @Bind("status") TaskStatus status,
         @Bind("resolved") boolean resolved,
         @Bind("type") ThreadType type);
 
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT json FROM thread_entity WHERE updatedAt > :before AND taskStatus = :status AND "
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt > :before AND taskStatus = :status AND "
                 + "taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[]) "
                 + "ORDER BY updatedAt DESC "
                 + "LIMIT :limit",
         connectionType = POSTGRES)
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT json FROM thread_entity WHERE updatedAt > :before AND taskStatus = :status AND "
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt > :before AND taskStatus = :status AND "
                 + "JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) "
                 + "ORDER BY updatedAt DESC "
                 + "LIMIT :limit",
@@ -571,19 +578,21 @@ public interface CollectionDAO {
         @Bind("userTeamJsonMysql") String userTeamJsonMysql,
         @Bind("limit") int limit,
         @Bind("before") long before,
-        @Bind("status") String status);
+        @Bind("status") TaskStatus status);
 
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT json FROM thread_entity WHERE updatedAt < :after AND taskStatus = :status AND "
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt < :after "
-                + "taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[]) "
+                + "AND (:status IS NULL OR taskStatus = :status) "
+                + "AND taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[]) "
                 + "ORDER BY updatedAt DESC "
                 + "LIMIT :limit",
         connectionType = POSTGRES)
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT json FROM thread_entity WHERE updatedAt < :after AND taskStatus = :status AND "
+            "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt < :after "
-                + "JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) "
+                + "AND (:status IS NULL OR taskStatus = :status) "
+                + "AND JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) "
                 + "ORDER BY updatedAt DESC "
                 + "LIMIT :limit",
         connectionType = MYSQL)
@@ -592,47 +601,53 @@ public interface CollectionDAO {
         @Bind("userTeamJsonMysql") String userTeamJsonMysql,
         @Bind("limit") int limit,
         @Bind("after") long after,
-        @Bind("status") String status);
+        @Bind("status") TaskStatus status);
 
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT count(id) FROM thread_entity WHERE taskStatus = :status AND "
+            "SELECT count(id) FROM thread_entity WHERE type='Task' AND "
+                + "(:status IS NULL OR taskStatus = :status) AND "
                 + "taskAssignees @> ANY (ARRAY[<userTeamJsonPostgres>]::jsonb[])",
         connectionType = POSTGRES)
     @ConnectionAwareSqlQuery(
         value =
-            "SELECT count(id) FROM thread_entity WHERE taskStatus = :status AND "
+            "SELECT count(id) FROM thread_entity WHERE "
+                + "(:status IS NULL OR taskStatus = :status) AND "
                 + "JSON_OVERLAPS(taskAssignees, :userTeamJsonMysql) ",
         connectionType = MYSQL)
     int listCountTasksAssignedTo(
         @BindList("userTeamJsonPostgres") List<String> userTeamJsonPostgres,
         @Bind("userTeamJsonMysql") String userTeamJsonMysql,
-        @Bind("status") String status);
+        @Bind("status") TaskStatus status);
 
     @SqlQuery(
-        "SELECT json FROM thread_entity WHERE updatedAt > :before AND taskStatus = :status AND "
+        "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt > :before "
-            + "createdBy = :username "
+            + "AND (:status IS NULL OR taskStatus = :status) "
+            + "AND createdBy = :username "
             + "ORDER BY updatedAt DESC "
             + "LIMIT :limit")
     List<String> listTasksAssignedByBefore(
         @Bind("username") String username,
         @Bind("limit") int limit,
         @Bind("before") long before,
-        @Bind("status") String status);
+        @Bind("status") TaskStatus status);
 
     @SqlQuery(
-        "SELECT json FROM thread_entity WHERE updatedAt < :after AND taskStatus = :status AND "
+        "SELECT json FROM thread_entity WHERE type='Task' AND updatedAt < :after "
-            + "createdBy = :username "
+            + "AND (:status IS NULL OR taskStatus = :status) "
+            + "AND createdBy = :username "
             + "ORDER BY updatedAt DESC "
             + "LIMIT :limit")
     List<String> listTasksAssignedByAfter(
         @Bind("username") String username,
         @Bind("limit") int limit,
         @Bind("after") long after,
-        @Bind("status") String status);
+        @Bind("status") TaskStatus status);
 
-    @SqlQuery("SELECT count(id) FROM thread_entity WHERE taskStatus = :status AND " + "createdBy = :username")
+    @SqlQuery(
-    int listCountTasksAssignedBy(@Bind("username") String username, @Bind("status") String status);
+        "SELECT count(id) FROM thread_entity WHERE type='Task' "
+            + "AND (:status IS NULL OR taskStatus = :status) AND createdBy = :username")
+    int listCountTasksAssignedBy(@Bind("username") String username, @Bind("status") TaskStatus status);
 
     @SqlQuery(
         "SELECT json FROM thread_entity WHERE updatedAt > :before AND resolved = :resolved AND (:type IS NULL OR type = :type) AND "
@@ -679,10 +694,11 @@ public interface CollectionDAO {
         @Bind("resolved") boolean resolved);
 
     @SqlQuery(
-        "SELECT json FROM thread_entity WHERE updatedAt > :before AND resolved = :resolved AND (:type IS NULL OR type = :type) AND "
+        "SELECT json FROM thread_entity WHERE updatedAt > :before AND resolved = :resolved "
-            + "id in (SELECT fromFQN FROM field_relationship WHERE "
+            + "AND (:type IS NULL OR type = :type) "
-            + "(toFQN LIKE CONCAT(:fqnPrefix, '.%') OR toFQN=:fqnPrefix) AND fromType='THREAD' AND "
+            + "AND id in (SELECT fromFQN FROM field_relationship WHERE "
-            + "(toType LIKE CONCAT(:toType, '.%') OR toType=:toType) AND relation= :relation) "
+            + "(:fqnPrefix IS NULL OR toFQN LIKE CONCAT(:fqnPrefix, '.%') OR toFQN=:fqnPrefix) AND fromType='THREAD' AND "
+            + "(:toType IS NULL OR toType LIKE CONCAT(:toType, '.%') OR toType=:toType) AND relation= :relation) "
             + "ORDER BY updatedAt DESC "
             + "LIMIT :limit")
     List<String> listThreadsByEntityLinkBefore(
@@ -691,15 +707,17 @@ public interface CollectionDAO {
         @Bind("limit") int limit,
         @Bind("before") long before,
         @Bind("type") ThreadType type,
+        @Bind("status") TaskStatus status,
         @Bind("resolved") boolean resolved,
         @Bind("relation") int relation);
 
     @SqlQuery(
-        "SELECT json FROM thread_entity WHERE updatedAt < :after AND resolved = :resolved AND "
+        "SELECT json FROM thread_entity WHERE updatedAt < :after AND resolved = :resolved "
-            + "(:type IS NULL OR type = :type) AND "
+            + "AND (:status IS NULL OR taskStatus = :status) "
-            + "id in (SELECT fromFQN FROM field_relationship WHERE "
+            + "AND (:type IS NULL OR type = :type) "
-            + "(toFQN LIKE CONCAT(:fqnPrefix, '.%') OR toFQN=:fqnPrefix) AND fromType='THREAD' AND "
+            + "AND id in (SELECT fromFQN FROM field_relationship WHERE "
-            + "(toType LIKE CONCAT(:toType, '.%') OR toType=:toType) AND relation= :relation) "
+            + "(:fqnPrefix IS NULL OR toFQN LIKE CONCAT(:fqnPrefix, '.%') OR toFQN=:fqnPrefix) AND fromType='THREAD' AND "
+            + "(:toType IS NULL OR toType LIKE CONCAT(:toType, '.%') OR toType=:toType) AND relation= :relation) "
             + "ORDER BY updatedAt DESC "
             + "LIMIT :limit")
     List<String> listThreadsByEntityLinkAfter(
@@ -708,19 +726,22 @@ public interface CollectionDAO {
         @Bind("limit") int limit,
         @Bind("after") long after,
         @Bind("type") ThreadType type,
+        @Bind("status") TaskStatus status,
         @Bind("resolved") boolean resolved,
         @Bind("relation") int relation);
 
     @SqlQuery(
-        "SELECT count(id) FROM thread_entity WHERE resolved = :resolved AND "
+        "SELECT count(id) FROM thread_entity WHERE resolved = :resolved "
-            + "(:type IS NULL OR type = :type) AND "
+            + "AND (:status IS NULL OR taskStatus = :status) "
-            + "id in (SELECT fromFQN FROM field_relationship WHERE "
+            + "AND (:type IS NULL OR type = :type) "
-            + "(toFQN LIKE CONCAT(:fqnPrefix, '.%') OR toFQN=:fqnPrefix) AND fromType='THREAD' AND "
+            + "AND id in (SELECT fromFQN FROM field_relationship WHERE "
-            + "(toType LIKE CONCAT(:toType, '.%') OR toType=:toType) AND relation= :relation)")
+            + "(:fqnPrefix IS NULL OR toFQN LIKE CONCAT(:fqnPrefix, '.%') OR toFQN=:fqnPrefix) AND fromType='THREAD' AND "
+            + "(:toType IS NULL OR toType LIKE CONCAT(:toType, '.%') OR toType=:toType) AND relation= :relation)")
     int listCountThreadsByEntityLink(
         @Bind("fqnPrefix") String fqnPrefix,
         @Bind("toType") String toType,
         @Bind("type") ThreadType type,
+        @Bind("status") TaskStatus status,
         @Bind("resolved") boolean resolved,
         @Bind("relation") int relation);
 
@@ -732,9 +753,10 @@ public interface CollectionDAO {
 
     @SqlQuery(
         "SELECT entityLink, COUNT(id) count FROM field_relationship fr INNER JOIN thread_entity te ON fr.fromFQN=te.id "
-            + "WHERE (fr.toFQN LIKE CONCAT(:fqnPrefix, '.%') OR fr.toFQN=:fqnPrefix) AND "
+            + "WHERE (:fqnPrefix IS NULL OR fr.toFQN LIKE CONCAT(:fqnPrefix, '.%') OR fr.toFQN=:fqnPrefix) AND "
-            + "(fr.toType like concat(:toType, '.%') OR fr.toType=:toType)AND fr.fromType = :fromType "
+            + "(:toType IS NULL OR fr.toType like concat(:toType, '.%') OR fr.toType=:toType) AND fr.fromType = :fromType "
-            + "AND fr.relation = :relation AND te.resolved= :isResolved AND (:type IS NULL OR te.type = :type) "
+            + "AND fr.relation = :relation AND te.resolved= :isResolved AND (:status IS NULL OR te.taskStatus = :status) "
+            + "AND (:type IS NULL OR te.type = :type) "
             + "GROUP BY entityLink")
     @RegisterRowMapper(CountFieldMapper.class)
     List<List<String>> listCountByEntityLink(
@@ -743,6 +765,7 @@ public interface CollectionDAO {
         @Bind("toType") String toType,
         @Bind("relation") int relation,
         @Bind("type") ThreadType type,
+        @Bind("status") TaskStatus status,
         @Bind("isResolved") boolean isResolved);
 
     @SqlQuery(
@@ -761,11 +784,13 @@ public interface CollectionDAO {
 
     @SqlQuery(
         "SELECT entityLink, COUNT(id) count FROM thread_entity WHERE (id IN (<threadIds>)) "
-            + "AND resolved= :isResolved AND (:type IS NULL OR type = :type) GROUP BY entityLink")
+            + "AND resolved= :isResolved AND (:type IS NULL OR type = :type) "
+            + "AND (:status IS NULL OR taskStatus = :status) GROUP BY entityLink")
     @RegisterRowMapper(CountFieldMapper.class)
     List<List<String>> listCountByThreads(
         @BindList("threadIds") List<String> threadIds,
         @Bind("type") ThreadType type,
+        @Bind("status") TaskStatus status,
         @Bind("isResolved") boolean isResolved);
 
     @SqlQuery(

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

@@ -204,32 +204,42 @@ public class FeedRepository {
         case TABLE:
           Table table = JsonUtils.readValue(json, Table.class);
           String oldJson = JsonUtils.pojoToJson(table);
-          if (entityLink.getFieldName().equals("columns")) {
+          if (entityLink.getFieldName() != null) {
-            Optional<Column> col =
+            if (entityLink.getFieldName().equals("columns")) {
-                table.getColumns().stream().filter(c -> c.getName().equals(entityLink.getArrayFieldName())).findFirst();
+              Optional<Column> col =
-            if (col.isPresent()) {
+                  table.getColumns().stream()
-              Column column = col.get();
+                      .filter(c -> c.getName().equals(entityLink.getArrayFieldName()))
-              if (descriptionTasks.contains(taskType)) {
+                      .findFirst();
-                column.setDescription(newValue);
+              if (col.isPresent()) {
-              } else if (tagTasks.contains(taskType)) {
+                Column column = col.get();
-                List<TagLabel> tags = JsonUtils.readObjects(newValue, TagLabel.class);
+                if (descriptionTasks.contains(taskType)) {
-                column.setTags(tags);
+                  column.setDescription(newValue);
+                } else if (tagTasks.contains(taskType)) {
+                  List<TagLabel> tags = JsonUtils.readObjects(newValue, TagLabel.class);
+                  column.setTags(tags);
+                }
+              } else {
+                throw new IllegalArgumentException(
+                    String.format(
+                        "The Column with name '%s' is not found in the table.", entityLink.getArrayFieldName()));
               }
+            } else if (descriptionTasks.contains(taskType) && entityLink.getFieldName().equals("description")) {
+              table.setDescription(newValue);
+            } else if (tagTasks.contains(taskType) && entityLink.getFieldName().equals("tags")) {
+              List<TagLabel> tags = JsonUtils.readObjects(newValue, TagLabel.class);
+              table.setTags(tags);
             } else {
+              // Not supported
               throw new IllegalArgumentException(
                   String.format(
-                      "The Column with name '%s' is not found in the table.", entityLink.getArrayFieldName()));
+                      "The field name %s is not supported for %s task.", entityLink.getFieldName(), task.getType()));
             }
-          } else if (descriptionTasks.contains(taskType) && entityLink.getFieldName().equals("description")) {
-            table.setDescription(newValue);
-          } else if (tagTasks.contains(taskType) && entityLink.getFieldName().equals("tags")) {
-            List<TagLabel> tags = JsonUtils.readObjects(newValue, TagLabel.class);
-            table.setTags(tags);
           } else {
             // Not supported
             throw new IllegalArgumentException(
                 String.format(
-                    "The field name %s is not supported for %s task.", entityLink.getFieldName(), task.getType()));
+                    "The Entity link with no field name - %s is not supported for %s task.",
+                    entityLink, task.getType()));
           }
           String updatedEntityJson = JsonUtils.pojoToJson(table);
           JsonPatch patch = JsonUtils.getJsonPatch(oldJson, updatedEntityJson);
@@ -479,7 +489,7 @@ public class FeedRepository {
   }
 
   @Transaction
-  public ThreadCount getThreadsCount(String link, ThreadType type, boolean isResolved) {
+  public ThreadCount getThreadsCount(String link, ThreadType type, TaskStatus taskStatus, boolean isResolved) {
     ThreadCount threadCount = new ThreadCount();
     List<List<String>> result;
     List<EntityLinkThreadCount> entityLinkThreadCounts = new ArrayList<>();
@@ -488,8 +498,7 @@ public class FeedRepository {
       // Get thread count of all entities
       result =
           dao.feedDAO()
-              .listCountByEntityLink(
+              .listCountByEntityLink(null, Entity.THREAD, null, IS_ABOUT.ordinal(), type, taskStatus, isResolved);
-                  StringUtils.EMPTY, Entity.THREAD, StringUtils.EMPTY, IS_ABOUT.ordinal(), type, isResolved);
     } else {
       EntityLink entityLink = EntityLink.parse(link);
       EntityReference reference = EntityUtil.validateEntityLink(entityLink);
@@ -511,6 +520,7 @@ public class FeedRepository {
                     entityLink.getFullyQualifiedFieldType(),
                     IS_ABOUT.ordinal(),
                     type,
+                    taskStatus,
                     isResolved);
       }
     }
@@ -560,12 +570,12 @@ public class FeedRepository {
       // Get one extra result used for computing before cursor
       List<String> jsons;
       if (paginationType == PaginationType.BEFORE) {
-        jsons = dao.feedDAO().listBefore(limit + 1, time, isResolved, threadType);
+        jsons = dao.feedDAO().listBefore(limit + 1, time, taskStatus, isResolved, threadType);
       } else {
-        jsons = dao.feedDAO().listAfter(limit + 1, time, isResolved, threadType);
+        jsons = dao.feedDAO().listAfter(limit + 1, time, taskStatus, isResolved, threadType);
       }
       threads = JsonUtils.readObjects(jsons, Thread.class);
-      total = dao.feedDAO().listCount(isResolved, threadType);
+      total = dao.feedDAO().listCount(taskStatus, isResolved, threadType);
     } else {
       // Either one or both the filters are enabled
       // we don't support both the filters together. If both are not null, entity link takes precedence
@@ -592,6 +602,7 @@ public class FeedRepository {
                         limit + 1,
                         time,
                         threadType,
+                        taskStatus,
                         isResolved,
                         IS_ABOUT.ordinal());
           } else {
@@ -603,6 +614,7 @@ public class FeedRepository {
                         limit + 1,
                         time,
                         threadType,
+                        taskStatus,
                         isResolved,
                         IS_ABOUT.ordinal());
           }
@@ -613,6 +625,7 @@ public class FeedRepository {
                       entityLink.getFullyQualifiedFieldValue(),
                       entityLink.getFullyQualifiedFieldType(),
                       threadType,
+                      taskStatus,
                       isResolved,
                       IS_ABOUT.ordinal());
         }
@@ -853,16 +866,12 @@ public class FeedRepository {
     List<String> userTeamJsonPostgres = getUserTeamJsonPostgres(userId, teamIds);
     String userTeamJsonMysql = getUserTeamJsonMysql(userId, teamIds);
     if (paginationType == PaginationType.BEFORE) {
-      jsons =
+      jsons = dao.feedDAO().listTasksAssignedToBefore(userTeamJsonPostgres, userTeamJsonMysql, limit, time, status);
-          dao.feedDAO()
-              .listTasksAssignedToBefore(userTeamJsonPostgres, userTeamJsonMysql, limit, time, status.toString());
     } else {
-      jsons =
+      jsons = dao.feedDAO().listTasksAssignedToAfter(userTeamJsonPostgres, userTeamJsonMysql, limit, time, status);
-          dao.feedDAO()
-              .listTasksAssignedToAfter(userTeamJsonPostgres, userTeamJsonMysql, limit, time, status.toString());
     }
     List<Thread> threads = JsonUtils.readObjects(jsons, Thread.class);
-    int totalCount = dao.feedDAO().listCountTasksAssignedTo(userTeamJsonPostgres, userTeamJsonMysql, status.toString());
+    int totalCount = dao.feedDAO().listCountTasksAssignedTo(userTeamJsonPostgres, userTeamJsonMysql, status);
     sortPostsInThreads(threads);
     return new FilteredThreads(threads, totalCount);
   }
@@ -891,12 +900,12 @@ public class FeedRepository {
     String username = user.getName();
     List<String> jsons;
     if (paginationType == PaginationType.BEFORE) {
-      jsons = dao.feedDAO().listTasksAssignedByBefore(username, limit, time, status.toString());
+      jsons = dao.feedDAO().listTasksAssignedByBefore(username, limit, time, status);
     } else {
-      jsons = dao.feedDAO().listTasksAssignedByAfter(username, limit, time, status.toString());
+      jsons = dao.feedDAO().listTasksAssignedByAfter(username, limit, time, status);
     }
     List<Thread> threads = JsonUtils.readObjects(jsons, Thread.class);
-    int totalCount = dao.feedDAO().listCountTasksAssignedBy(username, status.toString());
+    int totalCount = dao.feedDAO().listCountTasksAssignedBy(username, status);
     sortPostsInThreads(threads);
     return new FilteredThreads(threads, totalCount);
   }

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/feeds/FeedResource.java

@@ -186,9 +186,8 @@ public class FeedResource {
           ThreadType threadType,
       @Parameter(
               description =
-                  "The status of tasks to filter the results. It can take one of 'Open', 'Closed'. This filter will take effect only when threadType is set to Task",
+                  "The status of tasks to filter the results. It can take one of 'Open', 'Closed'. This filter will take effect only when type is set to Task",
               schema = @Schema(implementation = TaskStatus.class))
-          @DefaultValue("Open")
           @QueryParam("taskStatus")
           TaskStatus taskStatus)
       throws IOException {
@@ -364,11 +363,17 @@ public class FeedResource {
               schema = @Schema(implementation = ThreadType.class))
           @QueryParam("type")
           ThreadType threadType,
+      @Parameter(
+              description =
+                  "The status of tasks to filter the results. It can take one of 'Open', 'Closed'. This filter will take effect only when type is set to Task",
+              schema = @Schema(implementation = TaskStatus.class))
+          @QueryParam("taskStatus")
+          TaskStatus taskStatus,
       @Parameter(description = "Filter threads by whether it is active or resolved", schema = @Schema(type = "boolean"))
           @DefaultValue("false")
           @QueryParam("isResolved")
           Boolean isResolved) {
-    return dao.getThreadsCount(entityLink, threadType, isResolved);
+    return dao.getThreadsCount(entityLink, threadType, taskStatus, isResolved);
   }
 
   @POST

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

@@ -20,6 +20,7 @@ import static javax.ws.rs.core.Response.Status.OK;
 import static org.awaitility.Awaitility.with;
 import static org.awaitility.Durations.ONE_SECOND;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -310,13 +311,13 @@ public class FeedResourceTest extends CatalogApplicationTest {
 
   @Test
   void post_validTaskAndList_200() throws IOException {
-    int totalTaskCount = listTasks(null, null, null, null, ADMIN_AUTH_HEADERS).getPaging().getTotal();
+    int totalTaskCount = listTasks(null, null, null, null, null, ADMIN_AUTH_HEADERS).getPaging().getTotal();
     int assignedByCount =
-        listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_BY.toString(), null, ADMIN_AUTH_HEADERS)
+        listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_BY.toString(), null, null, ADMIN_AUTH_HEADERS)
             .getPaging()
             .getTotal();
     int assignedToCount =
-        listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_TO.toString(), null, ADMIN_AUTH_HEADERS)
+        listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_TO.toString(), null, null, ADMIN_AUTH_HEADERS)
             .getPaging()
             .getTotal();
     CreateTaskDetails taskDetails =
@@ -330,7 +331,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
 
     Map<String, String> userAuthHeaders = authHeaders(USER.getEmail());
     createAndCheck(create, userAuthHeaders);
-    ThreadList tasks = listTasks(null, null, null, null, userAuthHeaders);
+    ThreadList tasks = listTasks(null, null, null, null, null, userAuthHeaders);
     TaskDetails task = tasks.getData().get(0).getTask();
     assertNotNull(task.getId());
     int task1Id = task.getId();
@@ -356,6 +357,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
             .withAssignees(List.of(USER.getEntityReference()))
             .withType(TaskType.RequestDescription)
             .withSuggestion("new description2");
+    about = about.substring(0, about.length() - 1) + "::columns::c1::description>";
     create =
         new CreateThread()
             .withAbout(about)
@@ -364,7 +366,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
             .withMessage("Request Description for " + TABLE2.getName())
             .withTaskDetails(taskDetails);
     createAndCheck(create, userAuthHeaders);
-    tasks = listTasks(null, null, null, null, userAuthHeaders);
+    tasks = listTasks(null, null, null, null, null, userAuthHeaders);
     task = tasks.getData().get(0).getTask();
     assertNotNull(task.getId());
     int task2Id = task.getId();
@@ -376,20 +378,38 @@ public class FeedResourceTest extends CatalogApplicationTest {
     assertEquals(totalTaskCount + 2, tasks.getData().size());
 
     // try to list tasks with filters
-    tasks = listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_BY.toString(), null, userAuthHeaders);
+    tasks = listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_BY.toString(), null, null, userAuthHeaders);
     task = tasks.getData().get(0).getTask();
     assertEquals(task1Id, task.getId());
     assertEquals("new description", task.getSuggestion());
     assertEquals(assignedByCount + 1, tasks.getPaging().getTotal());
     assertEquals(assignedByCount + 1, tasks.getData().size());
 
-    tasks = listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_TO.toString(), null, userAuthHeaders);
+    tasks = listTasks(null, USER.getId().toString(), FilterType.ASSIGNED_TO.toString(), null, null, userAuthHeaders);
     task = tasks.getData().get(0).getTask();
     assertEquals(task2Id, task.getId());
     assertEquals(USER.getFullyQualifiedName(), task.getAssignees().get(0).getFullyQualifiedName());
     assertEquals("new description2", task.getSuggestion());
     assertEquals(assignedToCount + 1, tasks.getPaging().getTotal());
     assertEquals(assignedToCount + 1, tasks.getData().size());
+
+    ThreadCount count = listTasksCount(null, TaskStatus.Open, userAuthHeaders);
+    int totalOpenTaskCount = count.getTotalCount();
+    count = listTasksCount(null, TaskStatus.Closed, userAuthHeaders);
+    int totalClosedTaskCount = count.getTotalCount();
+
+    // close a task and test the task status filter
+    ResolveTask resolveTask = new ResolveTask().withNewValue("accepted description");
+    resolveTask(task2Id, resolveTask, userAuthHeaders);
+
+    tasks = listTasks(null, null, null, TaskStatus.Open, null, userAuthHeaders);
+    assertFalse(tasks.getData().stream().anyMatch(t -> t.getTask().getId().equals(task2Id)));
+    assertEquals(totalOpenTaskCount - 1, tasks.getPaging().getTotal());
+
+    tasks = listTasks(null, null, null, TaskStatus.Closed, null, userAuthHeaders);
+    assertEquals(task2Id, tasks.getData().get(0).getTask().getId());
+    assertEquals(totalClosedTaskCount + 1, tasks.getPaging().getTotal());
+    assertEquals(totalClosedTaskCount + 1, tasks.getData().size());
   }
 
   @Test
@@ -413,7 +433,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
     Map<String, String> userAuthHeaders = authHeaders(USER.getEmail());
     createAndCheck(create, userAuthHeaders);
 
-    ThreadList tasks = listTasks(null, null, null, null, userAuthHeaders);
+    ThreadList tasks = listTasks(null, null, null, null, null, userAuthHeaders);
     TaskDetails task = tasks.getData().get(0).getTask();
     assertNotNull(task.getId());
     int taskId = task.getId();
@@ -520,7 +540,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
     Map<String, String> userAuthHeaders = authHeaders(USER.getEmail());
     createAndCheck(create, userAuthHeaders);
 
-    ThreadList tasks = listTasks(null, null, null, null, userAuthHeaders);
+    ThreadList tasks = listTasks(null, null, null, null, null, userAuthHeaders);
     TaskDetails task = tasks.getData().get(0).getTask();
     assertNotNull(task.getId());
     int taskId = task.getId();
@@ -584,7 +604,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
     // Get the first page
     ThreadList threads =
         listThreads(
-            entityLink, null, userAuthHeaders, null, null, ThreadType.Conversation.toString(), limit, null, null);
+            entityLink, null, userAuthHeaders, null, null, null, ThreadType.Conversation.toString(), limit, null, null);
     assertEquals(limit, threads.getData().size());
     assertEquals(totalThreadCount, threads.getPaging().getTotal());
     assertNotNull(threads.getPaging().getAfter());
@@ -602,6 +622,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
               userAuthHeaders,
               null,
               null,
+              null,
               ThreadType.Conversation.toString(),
               limit,
               null,
@@ -624,6 +645,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
             userAuthHeaders,
             null,
             null,
+            null,
             ThreadType.Conversation.toString(),
             limit,
             null,
@@ -639,6 +661,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
             userAuthHeaders,
             null,
             null,
+            null,
             ThreadType.Conversation.toString(),
             limit,
             beforeCursor,
@@ -1074,16 +1097,30 @@ public class FeedResourceTest extends CatalogApplicationTest {
   }
 
   public static ThreadList listTasks(
-      String entityLink, String userId, String filterType, Integer limitPosts, Map<String, String> authHeaders)
+      String entityLink,
+      String userId,
+      String filterType,
+      TaskStatus taskStatus,
+      Integer limitPosts,
+      Map<String, String> authHeaders)
       throws HttpResponseException {
     return listThreads(
-        entityLink, limitPosts, authHeaders, userId, filterType, ThreadType.Task.toString(), null, null, null);
+        entityLink,
+        limitPosts,
+        authHeaders,
+        userId,
+        filterType,
+        taskStatus,
+        ThreadType.Task.toString(),
+        null,
+        null,
+        null);
   }
 
   public static ThreadList listThreads(String entityLink, Integer limitPosts, Map<String, String> authHeaders)
       throws HttpResponseException {
     return listThreads(
-        entityLink, limitPosts, authHeaders, null, null, ThreadType.Conversation.toString(), null, null, null);
+        entityLink, limitPosts, authHeaders, null, null, null, ThreadType.Conversation.toString(), null, null, null);
   }
 
   public static ThreadList listThreads(
@@ -1092,6 +1129,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
       Map<String, String> authHeaders,
       String userId,
       String filterType,
+      TaskStatus taskStatus,
       String threadType,
       Integer limitParam,
       String before,
@@ -1100,6 +1138,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
     WebTarget target = getResource("feed");
     target = userId != null ? target.queryParam("userId", userId) : target;
     target = filterType != null ? target.queryParam("filterType", filterType) : target;
+    target = taskStatus != null ? target.queryParam("taskStatus", taskStatus) : target;
     target = threadType != null ? target.queryParam("type", threadType) : target;
     target = entityLink != null ? target.queryParam("entityLink", entityLink) : target;
     target = limitPosts != null ? target.queryParam("limitPosts", limitPosts) : target;
@@ -1137,6 +1176,15 @@ public class FeedResourceTest extends CatalogApplicationTest {
     return TestUtils.get(target, ThreadCount.class, authHeaders);
   }
 
+  public static ThreadCount listTasksCount(String entityLink, TaskStatus taskStatus, Map<String, String> authHeaders)
+      throws HttpResponseException {
+    WebTarget target = getResource("feed/count");
+    target = entityLink != null ? target.queryParam("entityLink", entityLink) : target;
+    target = target.queryParam("type", ThreadType.Task);
+    target = taskStatus != null ? target.queryParam("taskStatus", taskStatus) : target;
+    return TestUtils.get(target, ThreadCount.class, authHeaders);
+  }
+
   private int getThreadCount(String entityLink, Map<String, String> authHeaders) throws HttpResponseException {
     List<EntityLinkThreadCount> linkThreadCount = listThreadsCount(entityLink, authHeaders).getCounts();
     EntityLinkThreadCount threadCount =