Fix #2865: Thread should have its own message and posts should only be the replies (#2944)

catalog-rest-service/src/main/java/org/openmetadata/catalog/events/ChangeEventHandler.java

@@ -18,13 +18,11 @@ import static org.openmetadata.catalog.type.EventType.ENTITY_SOFT_DELETED;
 import static org.openmetadata.catalog.type.EventType.ENTITY_UPDATED;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
 import javax.ws.rs.container.ContainerRequestContext;
 import javax.ws.rs.container.ContainerResponseContext;
 import javax.ws.rs.core.Response.Status;
-import javax.ws.rs.core.SecurityContext;
 import lombok.extern.slf4j.Slf4j;
 import org.jdbi.v3.core.Jdbi;
 import org.openmetadata.catalog.CatalogApplicationConfig;
@@ -37,7 +35,6 @@ import org.openmetadata.catalog.type.ChangeEvent;
 import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.EventType;
 import org.openmetadata.catalog.type.FieldChange;
-import org.openmetadata.catalog.type.Post;
 import org.openmetadata.catalog.util.EntityInterface;
 import org.openmetadata.catalog.util.JsonUtils;
 import org.openmetadata.catalog.util.RestUtil;
@@ -54,7 +51,6 @@ public class ChangeEventHandler implements EventHandler {
 
   public Void process(ContainerRequestContext requestContext, ContainerResponseContext responseContext) {
     String method = requestContext.getMethod();
-    SecurityContext securityContext = requestContext.getSecurityContext();
     try {
       ChangeEvent changeEvent = getChangeEvent(method, responseContext);
       if (changeEvent != null) {
@@ -77,7 +73,7 @@ public class ChangeEventHandler implements EventHandler {
         List<Thread> threads = getThreads(responseContext);
         if (threads != null) {
           for (var thread : threads) {
-            feedDao.create(thread, securityContext);
+            feedDao.create(thread);
           }
         }
       }
@@ -240,12 +236,6 @@ public class ChangeEventHandler implements EventHandler {
           break;
       }
 
-      Post post =
-          new Post()
-              .withFrom(entityInterface.getUpdatedBy())
-              .withMessage(message)
-              .withPostTs(System.currentTimeMillis());
-
       threads.add(
           new Thread()
               .withId(UUID.randomUUID())
@@ -254,7 +244,7 @@ public class ChangeEventHandler implements EventHandler {
               .withAbout(link.getLinkString())
               .withUpdatedBy(entityInterface.getUpdatedBy())
               .withUpdatedAt(System.currentTimeMillis())
-              .withPosts(Collections.singletonList(post)));
+              .withMessage(message));
     }
 
     return threads;

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

@@ -24,7 +24,6 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
-import javax.ws.rs.core.SecurityContext;
 import org.apache.commons.lang3.StringUtils;
 import org.jdbi.v3.sqlobject.transaction.Transaction;
 import org.openmetadata.catalog.Entity;
@@ -34,7 +33,6 @@ import org.openmetadata.catalog.entity.feed.Thread;
 import org.openmetadata.catalog.resources.feeds.FeedUtil;
 import org.openmetadata.catalog.resources.feeds.MessageParser;
 import org.openmetadata.catalog.resources.feeds.MessageParser.EntityLink;
-import org.openmetadata.catalog.security.SecurityUtil;
 import org.openmetadata.catalog.type.EntityReference;
 import org.openmetadata.catalog.type.Include;
 import org.openmetadata.catalog.type.Post;
@@ -50,13 +48,11 @@ public class FeedRepository {
   }
 
   @Transaction
-  public Thread create(Thread thread, SecurityContext securityContext) throws IOException, ParseException {
+  public Thread create(Thread thread) throws IOException, ParseException {
-    String fromUser = thread.getPosts().get(0).getFrom();
+    String createdBy = thread.getCreatedBy();
 
-    if (SecurityUtil.isSecurityEnabled(securityContext)) {
+    // Validate user creating thread
-      // Validate user creating thread if security is enabled
+    dao.userDAO().findEntityByName(createdBy);
-      dao.userDAO().findEntityByName(fromUser);
-    }
 
     // Validate about data entity is valid
     EntityLink about = EntityLink.parse(thread.getAbout());
@@ -69,7 +65,8 @@ public class FeedRepository {
     dao.feedDAO().insert(JsonUtils.pojoToJson(thread));
 
     // Add relationship User -- created --> Thread relationship
-    dao.relationshipDAO().insert(fromUser, thread.getId().toString(), "user", "thread", Relationship.CREATED.ordinal());
+    dao.relationshipDAO()
+        .insert(createdBy, thread.getId().toString(), "user", "thread", Relationship.CREATED.ordinal());
 
     // Add field relationship data asset Thread -- isAbout ---> entity/entityField
     // relationship
@@ -94,7 +91,7 @@ public class FeedRepository {
 
     // Create relationship for users, teams, and other entities that are mentioned in the post
     // Multiple mentions of the same entity is handled by taking distinct mentions
-    List<EntityLink> mentions = MessageParser.getEntityLinks(thread.getPosts().get(0).getMessage());
+    List<EntityLink> mentions = MessageParser.getEntityLinks(thread.getMessage());
 
     mentions.stream()
         .distinct()
@@ -117,7 +114,7 @@ public class FeedRepository {
 
   @Transaction
   public Thread addPostToThread(String id, Post post) throws IOException {
-    // Query 1 - validate user creating thread
+    // Query 1 - validate the user posting the message
     String fromUser = post.getFrom();
     dao.userDAO().findEntityByName(fromUser);
 
@@ -191,50 +188,57 @@ public class FeedRepository {
   }
 
   @Transaction
-  public List<Thread> listThreads(String link) throws IOException {
+  public List<Thread> listThreads(String link, int limitPosts) throws IOException {
+    List<Thread> threads = new ArrayList<>();
     if (link == null) {
       // Not listing thread by data asset or user
-      return JsonUtils.readObjects(dao.feedDAO().list(), Thread.class);
+      threads = JsonUtils.readObjects(dao.feedDAO().list(), Thread.class);
-    }
-    EntityLink entityLink = EntityLink.parse(link);
-    EntityReference reference = EntityUtil.validateEntityLink(entityLink);
-    List<String> threadIds = new ArrayList<>();
-    List<List<String>> result =
-        dao.fieldRelationshipDAO()
-            .listToByPrefix(
-                entityLink.getFullyQualifiedFieldValue(),
-                entityLink.getFullyQualifiedFieldType(),
-                "thread",
-                Relationship.MENTIONED_IN.ordinal());
-    result.forEach(l -> threadIds.add(l.get(1)));
-
-    // TODO remove hardcoding of thread
-    // For a user entitylink get created or replied relationships to the thread
-    if (reference.getType().equals(Entity.USER)) {
-      threadIds.addAll(getUserThreadIds(reference));
     } else {
-      // Only data assets are added as about
+      EntityLink entityLink = EntityLink.parse(link);
-      result =
+      EntityReference reference = EntityUtil.validateEntityLink(entityLink);
-          dao.fieldRelationshipDAO()
+      List<String> threadIds = new ArrayList<>();
-              .listFromByAllPrefix(
+      List<List<String>> result;
-                  entityLink.getFullyQualifiedFieldValue(),
-                  "thread",
-                  entityLink.getFullyQualifiedFieldType(),
-                  Relationship.IS_ABOUT.ordinal());
-      result.forEach(l -> threadIds.add(l.get(1)));
-    }
 
-    List<Thread> threads = new ArrayList<>();
+      // TODO remove hardcoding of thread
-    Set<String> uniqueValues = new HashSet<>();
+      // For a user entitylink get created or replied relationships to the thread
-    for (String t : threadIds) {
+      if (reference.getType().equals(Entity.USER)) {
-      // If an entity has multiple relationships (created, mentioned, repliedTo etc.) to the same thread
+        threadIds.addAll(getUserThreadIds(reference));
-      // Don't send duplicated copies of the thread in response
+      } else {
-      if (uniqueValues.add(t)) {
+        // Only data assets are added as about
-        threads.add(EntityUtil.validate(t, dao.feedDAO().findById(t), Thread.class));
+        result =
+            dao.fieldRelationshipDAO()
+                .listFromByAllPrefix(
+                    entityLink.getFullyQualifiedFieldValue(),
+                    "thread",
+                    entityLink.getFullyQualifiedFieldType(),
+                    Relationship.IS_ABOUT.ordinal());
+        result.forEach(l -> threadIds.add(l.get(1)));
+      }
+
+      Set<String> uniqueValues = new HashSet<>();
+      for (String t : threadIds) {
+        // If an entity has multiple relationships (created, mentioned, repliedTo etc.) to the same thread
+        // Don't send duplicated copies of the thread in response
+        if (uniqueValues.add(t)) {
+          threads.add(EntityUtil.validate(t, dao.feedDAO().findById(t), Thread.class));
+        }
+      }
+      // sort the list by thread updated timestamp before returning
+      threads.sort(Comparator.comparing(Thread::getUpdatedAt, Comparator.reverseOrder()));
+    }
+    return limitPostsInThreads(threads, limitPosts);
+  }
+
+  private List<Thread> limitPostsInThreads(List<Thread> threads, int limitPosts) {
+    for (Thread t : threads) {
+      List<Post> posts = t.getPosts();
+      if (posts.size() > limitPosts) {
+        // Only keep the last "n" number of posts
+        posts.sort(Comparator.comparing(Post::getPostTs));
+        posts = posts.subList(posts.size() - limitPosts, posts.size());
+        t.withPosts(posts);
       }
     }
-    // sort the list by thread updated timestamp before returning
-    threads.sort(Comparator.comparing(Thread::getUpdatedAt, Comparator.reverseOrder()));
     return threads;
   }
 

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

@@ -121,11 +121,13 @@ public class UserRepository extends EntityRepository<User> {
 
   @Override
   public User setFields(User user, Fields fields) throws IOException, ParseException {
+
     user.setProfile(fields.contains("profile") ? user.getProfile() : null);
     user.setTeams(fields.contains("teams") ? getTeams(user) : null);
     user.setRoles(fields.contains("roles") ? getRoles(user) : null);
     user.setOwns(fields.contains("owns") ? getOwns(user) : null);
     user.setFollows(fields.contains("follows") ? getFollows(user) : null);
+
     return user;
   }
 

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

@@ -27,6 +27,8 @@ import java.util.List;
 import java.util.Objects;
 import java.util.UUID;
 import javax.validation.Valid;
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
@@ -112,13 +114,21 @@ public class FeedResource {
       })
   public ThreadList list(
       @Context UriInfo uriInfo,
+      @Parameter(
+              description = "Limit the number of posts sorted by chronological order (1 to 1000000, default = 3)",
+              schema = @Schema(type = "integer"))
+          @Min(1)
+          @Max(1000000)
+          @DefaultValue("3")
+          @QueryParam("limitPosts")
+          int limitPosts,
       @Parameter(
               description = "Filter threads by entity link",
               schema = @Schema(type = "string", example = "<E#/{entityType}/{entityFQN}/{fieldName}>"))
           @QueryParam("entityLink")
           String entityLink)
       throws IOException {
-    return new ThreadList(addHref(uriInfo, dao.listThreads(entityLink)));
+    return new ThreadList(addHref(uriInfo, dao.listThreads(entityLink, limitPosts)));
   }
 
   @GET
@@ -180,8 +190,7 @@ public class FeedResource {
   public Response create(@Context UriInfo uriInfo, @Context SecurityContext securityContext, @Valid CreateThread create)
       throws IOException, ParseException {
     Thread thread = getThread(securityContext, create);
-    FeedUtil.addPost(thread, new Post().withMessage(create.getMessage()).withFrom(create.getFrom()));
+    addHref(uriInfo, dao.create(thread));
-    addHref(uriInfo, dao.create(thread, securityContext));
     return Response.created(thread.getHref()).entity(thread).build();
   }
 
@@ -223,7 +232,8 @@ public class FeedResource {
     return new Thread()
         .withId(UUID.randomUUID())
         .withThreadTs(System.currentTimeMillis())
-        .withCreatedBy(securityContext.getUserPrincipal().getName())
+        .withMessage(create.getMessage())
+        .withCreatedBy(create.getFrom())
         .withAbout(create.getAbout())
         .withAddressedTo(create.getAddressedTo())
         .withUpdatedBy(securityContext.getUserPrincipal().getName())

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

@@ -13,7 +13,6 @@
 
 package org.openmetadata.catalog.resources.feeds;
 
-import java.util.Collections;
 import org.openmetadata.catalog.entity.feed.Thread;
 import org.openmetadata.catalog.type.Post;
 
@@ -22,14 +21,9 @@ public final class FeedUtil {
   private FeedUtil() {}
 
   public static void addPost(Thread thread, Post post) {
-    if (thread.getPosts() == null || thread.getPosts().isEmpty()) {
+    // Add new post to the thread
-      // First post in the thread
+    post.setPostTs(System.currentTimeMillis());
-      post.setPostTs(thread.getThreadTs());
+    thread.getPosts().add(post);
-      thread.setPosts(Collections.singletonList(post));
+    thread.withPostsCount(thread.getPosts().size());
-    } else {
-      // Add new post to the thread
-      post.setPostTs(System.currentTimeMillis());
-      thread.getPosts().add(post);
-    }
   }
 }

catalog-rest-service/src/main/java/org/openmetadata/catalog/security/DefaultAuthorizer.java

@@ -248,11 +248,11 @@ public class DefaultAuthorizer implements Authorizer {
   private void addOrUpdateUser(User user) {
     try {
       RestUtil.PutResponse<User> addedUser = userRepository.createOrUpdate(null, user);
-      LOG.debug("Added admin user entry: {}", addedUser);
+      LOG.debug("Added user entry: {}", addedUser);
     } catch (IOException | ParseException exception) {
       // In HA set up the other server may have already added the user.
       LOG.debug("Caught exception: {}", ExceptionUtils.getStackTrace(exception));
-      LOG.debug("Admin user entry: {} already exists.", user);
+      LOG.debug("User entry: {} already exists.", user);
     }
   }
 }

catalog-rest-service/src/main/resources/json/schema/entity/feed/thread.json

@@ -65,6 +65,15 @@
       "type": "boolean",
       "default": false
     },
+    "message": {
+      "description": "The main message of the thread in markdown format",
+      "type": "string"
+    },
+    "postsCount": {
+      "description": "The total count of posts in the thread",
+      "type": "integer",
+      "default": 0
+    },
     "posts": {
       "type": "array",
       "items": {

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

@@ -33,6 +33,7 @@ import java.util.Map;
 import java.util.UUID;
 import javax.ws.rs.client.WebTarget;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.http.client.HttpResponseException;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.MethodOrderer;
@@ -138,7 +139,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
   }
 
   @Test
-  void post_feedWithNonExistentFrom_404() throws IOException {
+  void post_feedWithNonExistentFrom_404() {
     // Create thread with non-existent from
     CreateThread create = create().withFrom(NON_EXISTENT_ENTITY.toString());
     assertResponse(
@@ -154,7 +155,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
   }
 
   @Test
-  void post_feedWithInvalidAbout_400() throws IOException {
+  void post_feedWithInvalidAbout_400() {
     // post with invalid entity link pattern
     // if entity link refers to an array member, then it should have both
     // field name and value
@@ -165,12 +166,12 @@ public class FeedResourceTest extends CatalogApplicationTest {
 
   @Test
   void post_validThreadAndList_200(TestInfo test) throws IOException {
-    int totalThreadCount = listThreads(null, ADMIN_AUTH_HEADERS).getData().size();
+    int totalThreadCount = listThreads(null, null, ADMIN_AUTH_HEADERS).getData().size();
-    int userThreadCount = listThreads(USER_LINK, ADMIN_AUTH_HEADERS).getData().size();
+    int userThreadCount = listThreads(USER_LINK, null, ADMIN_AUTH_HEADERS).getData().size();
-    int teamThreadCount = listThreads(TEAM_LINK, ADMIN_AUTH_HEADERS).getData().size();
+    int teamThreadCount = listThreads(TEAM_LINK, null, ADMIN_AUTH_HEADERS).getData().size();
-    int tableThreadCount = listThreads(TABLE_LINK, ADMIN_AUTH_HEADERS).getData().size();
+    int tableThreadCount = listThreads(TABLE_LINK, null, ADMIN_AUTH_HEADERS).getData().size();
-    int tableDescriptionThreadCount = listThreads(TABLE_DESCRIPTION_LINK, ADMIN_AUTH_HEADERS).getData().size();
+    int tableDescriptionThreadCount = listThreads(TABLE_DESCRIPTION_LINK, null, ADMIN_AUTH_HEADERS).getData().size();
-    int tableColumnDescriptionThreadCount = listThreads(TABLE_COLUMN_LINK, ADMIN_AUTH_HEADERS).getData().size();
+    int tableColumnDescriptionThreadCount = listThreads(TABLE_COLUMN_LINK, null, ADMIN_AUTH_HEADERS).getData().size();
 
     CreateThread create =
         create()
@@ -188,27 +189,57 @@ public class FeedResourceTest extends CatalogApplicationTest {
     for (int i = 0; i < 10; i++) {
       createAndCheck(create, userAuthHeaders);
       // List all the threads and make sure the number of threads increased by 1
-      assertEquals(++userThreadCount, listThreads(USER_LINK, userAuthHeaders).getData().size()); // Mentioned user
+      assertEquals(++userThreadCount, listThreads(USER_LINK, null, userAuthHeaders).getData().size()); // Mentioned user
-      assertEquals(++teamThreadCount, listThreads(TEAM_LINK, userAuthHeaders).getData().size()); // Mentioned team
+      // TODO: There is no support for team mentions yet.
-      assertEquals(++tableThreadCount, listThreads(TABLE_LINK, userAuthHeaders).getData().size()); // About TABLE
+      // assertEquals(++teamThreadCount, listThreads(TEAM_LINK, null, userAuthHeaders).getData().size()); // Mentioned
+      // team
+      assertEquals(++tableThreadCount, listThreads(TABLE_LINK, null, userAuthHeaders).getData().size()); // About TABLE
+      assertEquals(++totalThreadCount, listThreads(null, null, userAuthHeaders).getData().size()); // Overall threads
+    }
+
+    // List threads should not include mentioned entities
+    // It should only include threads which are about the entity link
+    assertEquals(
+        tableDescriptionThreadCount,
+        listThreads(TABLE_DESCRIPTION_LINK, null, userAuthHeaders).getData().size()); // About TABLE Description
+    assertEquals(
+        tableColumnDescriptionThreadCount,
+        listThreads(TABLE_COLUMN_LINK, null, userAuthHeaders).getData().size()); // About TABLE Column Description
+
+    create.withAbout(TABLE_DESCRIPTION_LINK);
+    for (int i = 0; i < 10; i++) {
+      createAndCheck(create, userAuthHeaders);
+      // List all the threads and make sure the number of threads increased by 1
+      assertEquals(++userThreadCount, listThreads(USER_LINK, null, userAuthHeaders).getData().size()); // Mentioned user
+      assertEquals(++tableThreadCount, listThreads(TABLE_LINK, null, userAuthHeaders).getData().size()); // About TABLE
       assertEquals(
           ++tableDescriptionThreadCount,
-          listThreads(TABLE_DESCRIPTION_LINK, userAuthHeaders).getData().size()); // About TABLE Description
+          listThreads(TABLE_DESCRIPTION_LINK, null, userAuthHeaders).getData().size()); // About TABLE Description
+      assertEquals(++totalThreadCount, listThreads(null, null, userAuthHeaders).getData().size()); // Overall threads
+    }
+
+    create.withAbout(TABLE_COLUMN_LINK);
+    for (int i = 0; i < 10; i++) {
+      createAndCheck(create, userAuthHeaders);
+      // List all the threads and make sure the number of threads increased by 1
+      assertEquals(++userThreadCount, listThreads(USER_LINK, null, userAuthHeaders).getData().size()); // Mentioned user
+      assertEquals(++tableThreadCount, listThreads(TABLE_LINK, null, userAuthHeaders).getData().size()); // About TABLE
       assertEquals(
           ++tableColumnDescriptionThreadCount,
-          listThreads(TABLE_COLUMN_LINK, userAuthHeaders).getData().size()); // About TABLE Column Description
+          listThreads(TABLE_COLUMN_LINK, null, userAuthHeaders).getData().size()); // About TABLE Description
-      assertEquals(++totalThreadCount, listThreads(null, userAuthHeaders).getData().size()); // Overall threads
+      assertEquals(++totalThreadCount, listThreads(null, null, userAuthHeaders).getData().size()); // Overall threads
     }
 
     // Test the /api/v1/feed/count API
     assertEquals(userThreadCount, listThreadsCount(USER_LINK, userAuthHeaders).getTotalCount());
-    assertEquals(tableThreadCount, getThreadCount(TABLE_LINK, userAuthHeaders));
+    assertEquals(tableDescriptionThreadCount, getThreadCount(TABLE_DESCRIPTION_LINK, userAuthHeaders));
+    assertEquals(tableColumnDescriptionThreadCount, getThreadCount(TABLE_COLUMN_LINK, userAuthHeaders));
   }
 
   @Test
   void post_addPostWithoutMessage_4xx() {
     // Add post to a thread without message field
-    Post post = createPost().withMessage(null);
+    Post post = createPost(null).withMessage(null);
 
     assertResponseContains(
         () -> addPost(THREAD.getId(), post, AUTH_HEADERS), BAD_REQUEST, "[message must not be null]");
@@ -217,7 +248,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
   @Test
   void post_addPostWithoutFrom_4xx() {
     // Add post to a thread without from field
-    Post post = createPost().withFrom(null);
+    Post post = createPost(null).withFrom(null);
 
     assertResponseContains(() -> addPost(THREAD.getId(), post, AUTH_HEADERS), BAD_REQUEST, "[from must not be null]");
   }
@@ -226,7 +257,7 @@ public class FeedResourceTest extends CatalogApplicationTest {
   void post_addPostWithNonExistentFrom_404() {
     // Add post to a thread with non-existent from user
 
-    Post post = createPost().withFrom(NON_EXISTENT_ENTITY.toString());
+    Post post = createPost(null).withFrom(NON_EXISTENT_ENTITY.toString());
     assertResponse(
         () -> addPost(THREAD.getId(), post, AUTH_HEADERS), NOT_FOUND, entityNotFound(Entity.USER, NON_EXISTENT_ENTITY));
   }
@@ -237,15 +268,51 @@ public class FeedResourceTest extends CatalogApplicationTest {
     // Add 10 posts and validate
     int POST_COUNT = 10;
     for (int i = 0; i < POST_COUNT; i++) {
-      Post post = createPost();
+      Post post = createPost(null);
       thread = addPostAndCheck(thread, post, AUTH_HEADERS);
     }
 
     // Check if get posts API returns all the posts
     PostList postList = listPosts(thread.getId().toString(), AUTH_HEADERS);
-    // Thread also has the first message as a post.
+    assertEquals(POST_COUNT, postList.getData().size());
-    // So, the total count should be POST_COUNT+1
+  }
-    assertEquals(POST_COUNT + 1, postList.getData().size());
+
+  @Test
+  void list_threadsWithPostsLimit() throws HttpResponseException {
+    Thread thread = createAndCheck(create(), AUTH_HEADERS);
+    // Add 10 posts and validate
+    int POST_COUNT = 10;
+    for (int i = 0; i < POST_COUNT; i++) {
+      Post post = createPost("message" + i);
+      thread = addPostAndCheck(thread, post, AUTH_HEADERS);
+    }
+
+    ThreadList threads = listThreads(null, 5, AUTH_HEADERS);
+    thread = threads.getData().get(0);
+    assertEquals(5, thread.getPosts().size());
+    assertEquals(POST_COUNT, thread.getPostsCount());
+    // Thread should contain the latest 5 messages
+    List<Post> posts = thread.getPosts();
+    int startIndex = 5;
+    for (var post : posts) {
+      assertEquals("message" + startIndex++, post.getMessage());
+    }
+
+    // when posts limit is null, it should return 3 posts which is the default
+    threads = listThreads(null, null, AUTH_HEADERS);
+    thread = threads.getData().get(0);
+    assertEquals(3, thread.getPosts().size());
+
+    // limit 0 is not supported and should throw an exception
+    assertResponse(
+        () -> listThreads(null, 0, AUTH_HEADERS),
+        BAD_REQUEST,
+        "[query param limitPosts must be greater than or equal to 1]");
+
+    // limit greater than total number of posts should return correct response
+    threads = listThreads(null, 100, AUTH_HEADERS);
+    thread = threads.getData().get(0);
+    assertEquals(10, thread.getPosts().size());
   }
 
   @Test
@@ -281,9 +348,8 @@ public class FeedResourceTest extends CatalogApplicationTest {
 
   private static void validateThread(Thread thread, String message, String from, String about) {
     assertNotNull(thread.getId());
-    Post firstPost = thread.getPosts().get(0);
+    assertEquals(message, thread.getMessage());
-    assertEquals(message, firstPost.getMessage());
+    assertEquals(from, thread.getCreatedBy());
-    assertEquals(from, firstPost.getFrom());
     assertEquals(about, thread.getAbout());
   }
 
@@ -311,8 +377,9 @@ public class FeedResourceTest extends CatalogApplicationTest {
     return new CreateThread().withFrom(USER.getName()).withMessage("message").withAbout(about);
   }
 
-  public static Post createPost() {
+  public static Post createPost(String message) {
-    return new Post().withFrom(USER.getName()).withMessage("message");
+    message = StringUtils.isNotEmpty(message) ? message : "message";
+    return new Post().withFrom(USER.getName()).withMessage(message);
   }
 
   public static Thread getThread(UUID id, Map<String, String> authHeaders) throws HttpResponseException {
@@ -320,10 +387,11 @@ public class FeedResourceTest extends CatalogApplicationTest {
     return TestUtils.get(target, Thread.class, authHeaders);
   }
 
-  public static ThreadList listThreads(String entityLink, Map<String, String> authHeaders)
+  public static ThreadList listThreads(String entityLink, Integer limitPosts, Map<String, String> authHeaders)
       throws HttpResponseException {
     WebTarget target = getResource("feed");
     target = entityLink != null ? target.queryParam("entityLink", entityLink) : target;
+    target = limitPosts != null ? target.queryParam("limitPosts", limitPosts) : target;
     return TestUtils.get(target, ThreadList.class, authHeaders);
   }