diff --git a/lms/djangoapps/discussion/rest_api/api.py b/lms/djangoapps/discussion/rest_api/api.py
index 2caa901d518..f79cd663718 100644
--- a/lms/djangoapps/discussion/rest_api/api.py
+++ b/lms/djangoapps/discussion/rest_api/api.py
@@ -127,8 +127,8 @@
     discussion_open_for_user,
     get_usernames_for_course,
     get_usernames_from_search_string,
-    is_posting_allowed,
     set_attribute,
+    is_posting_allowed
 )
 
 
diff --git a/lms/djangoapps/discussion/rest_api/discussions_notifications.py b/lms/djangoapps/discussion/rest_api/discussions_notifications.py
new file mode 100644
index 00000000000..f3a5e3d61d9
--- /dev/null
+++ b/lms/djangoapps/discussion/rest_api/discussions_notifications.py
@@ -0,0 +1,260 @@
+"""
+Discussion notifications sender util.
+"""
+from django.conf import settings
+from lms.djangoapps.discussion.django_comment_client.permissions import get_team
+from openedx_events.learning.data import UserNotificationData
+from openedx_events.learning.signals import USER_NOTIFICATION_REQUESTED
+
+from common.djangoapps.student.models import CourseEnrollment
+from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
+from openedx.core.djangoapps.course_groups.models import CourseCohortsSettings, CourseUserGroup
+from openedx.core.djangoapps.discussions.utils import get_divided_discussions
+from django.utils.translation import gettext_lazy as _
+
+from openedx.core.djangoapps.django_comment_common.comment_client.comment import Comment
+from openedx.core.djangoapps.django_comment_common.comment_client.subscriptions import Subscription
+from openedx.core.djangoapps.django_comment_common.models import (
+    FORUM_ROLE_ADMINISTRATOR,
+    FORUM_ROLE_COMMUNITY_TA,
+    FORUM_ROLE_MODERATOR,
+    CourseDiscussionSettings,
+    Role
+)
+
+
+class DiscussionNotificationSender:
+    """
+    Class to send notifications to users who are subscribed to the thread.
+    """
+
+    def __init__(self, thread, course, creator, parent_id=None):
+        self.thread = thread
+        self.course = course
+        self.creator = creator
+        self.parent_id = parent_id
+        self.parent_response = None
+        self._get_parent_response()
+
+    def _send_notification(self, user_ids, notification_type, extra_context=None):
+        """
+        Send notification to users
+        """
+        if not user_ids:
+            return
+
+        if extra_context is None:
+            extra_context = {}
+
+        notification_data = UserNotificationData(
+            user_ids=[int(user_id) for user_id in user_ids],
+            context={
+                "replier_name": self.creator.username,
+                "post_title": self.thread.title,
+                "course_name": self.course.display_name,
+                **extra_context,
+            },
+            notification_type=notification_type,
+            content_url=f"{settings.DISCUSSIONS_MICROFRONTEND_URL}/{str(self.course.id)}/posts/{self.thread.id}",
+            app_name="discussion",
+            course_key=self.course.id,
+        )
+        USER_NOTIFICATION_REQUESTED.send_event(notification_data=notification_data)
+
+    def _get_parent_response(self):
+        """
+        Get parent response object
+        """
+        if self.parent_id and not self.parent_response:
+            self.parent_response = Comment(id=self.parent_id).retrieve()
+
+        return self.parent_response
+
+    def send_new_response_notification(self):
+        """
+        Send notification to users who are subscribed to the main thread/post i.e.
+        there is a response to the main thread.
+        """
+        if not self.parent_id and self.creator.id != int(self.thread.user_id):
+            self._send_notification([self.thread.user_id], "new_response")
+
+    def _response_and_thread_has_same_creator(self) -> bool:
+        """
+        Check if response and main thread have same author.
+        """
+        return int(self.parent_response.user_id) == int(self.thread.user_id)
+
+    def _response_and_comment_has_same_creator(self):
+        return int(self.parent_response.attributes['user_id']) == self.creator.id
+
+    def send_new_comment_notification(self):
+        """
+        Send notification to parent thread creator i.e. comment on the response.
+        """
+        if (
+            self.parent_response and
+            self.creator.id != int(self.thread.user_id)
+        ):
+            # use your if author of response is same as author of post.
+            # use 'their' if comment author is also response author.
+            author_name = (
+                # Translators: Replier commented on "your" response to your post
+                _("your")
+                if self._response_and_thread_has_same_creator()
+                else (
+                    # Translators: Replier commented on "their" response to your post
+                    _("their")
+                    if self._response_and_comment_has_same_creator()
+                    else f"{self.parent_response.username}'s"
+                )
+            )
+            context = {
+                "author_name": str(author_name),
+            }
+            self._send_notification([self.thread.user_id], "new_comment", extra_context=context)
+
+    def send_new_comment_on_response_notification(self):
+        """
+        Send notification to parent response creator i.e. comment on the response.
+        Do not send notification if author of response is same as author of post.
+        """
+        if (
+            self.parent_response and
+            self.creator.id != int(self.parent_response.user_id) and not
+            self._response_and_thread_has_same_creator()
+        ):
+            self._send_notification([self.parent_response.user_id], "new_comment_on_response")
+
+    def _check_if_subscriber_is_not_thread_or_content_creator(self, subscriber_id) -> bool:
+        """
+        Check if the subscriber is not the thread creator or response creator
+        """
+        is_not_creator = (
+            subscriber_id != int(self.thread.user_id) and
+            subscriber_id != int(self.creator.id)
+        )
+        if self.parent_response:
+            return is_not_creator and subscriber_id != int(self.parent_response.user_id)
+
+        return is_not_creator
+
+    def send_response_on_followed_post_notification(self):
+        """
+        Send notification to followers of the thread/post
+        except:
+        Tread creator , response creator,
+        """
+        users = []
+        page = 1
+        has_more_subscribers = True
+
+        while has_more_subscribers:
+
+            subscribers = Subscription.fetch(self.thread.id, query_params={'page': page})
+            if page <= subscribers.num_pages:
+                for subscriber in subscribers.collection:
+                    # Check if the subscriber is not the thread creator or response creator
+                    subscriber_id = int(subscriber.get('subscriber_id'))
+                    # do not send notification to the user who created the response and the thread
+                    if self._check_if_subscriber_is_not_thread_or_content_creator(subscriber_id):
+                        users.append(subscriber_id)
+            else:
+                has_more_subscribers = False
+            page += 1
+        # Remove duplicate users from the list of users to send notification
+        users = list(set(users))
+        if not self.parent_id:
+            self._send_notification(users, "response_on_followed_post")
+        else:
+            self._send_notification(
+                users,
+                "comment_on_followed_post",
+                extra_context={"author_name": self.parent_response.username}
+            )
+
+    def _create_cohort_course_audience(self):
+        """
+        Creates audience based on user cohort and role
+        """
+        course_key_str = str(self.course.id)
+        discussion_cohorted = is_discussion_cohorted(course_key_str)
+
+        # Retrieves cohort divided discussion
+        discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
+        divided_course_wide_discussions, divided_inline_discussions = get_divided_discussions(
+            self.course,
+            discussion_settings
+        )
+
+        # Checks if post has any cohort assigned
+        group_id = self.thread.attributes['group_id']
+        if group_id is not None:
+            group_id = int(group_id)
+
+        # Course wide topics
+        topic_id = self.thread.attributes['commentable_id']
+        all_topics = divided_inline_discussions + divided_course_wide_discussions
+        topic_divided = topic_id in all_topics or discussion_settings.always_divide_inline_discussions
+
+        # Team object from topic id
+        team = get_team(topic_id)
+
+        user_ids = []
+        if team:
+            user_ids = team.users.all().values_list('id', flat=True)
+        elif discussion_cohorted and topic_divided and group_id is not None:
+            users_in_cohort = CourseUserGroup.objects.filter(
+                course_id=course_key_str, id=group_id
+            ).values_list('users__id', flat=True)
+            user_ids.extend(users_in_cohort)
+
+            privileged_roles = [FORUM_ROLE_ADMINISTRATOR, FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA]
+            privileged_users = Role.objects.filter(
+                name__in=privileged_roles,
+                course_id=course_key_str
+            ).values_list('users__id', flat=True)
+            user_ids.extend(privileged_users)
+
+            staff_users = CourseStaffRole(self.course.id).users_with_role().values_list('id', flat=True)
+            user_ids.extend(staff_users)
+
+            admin_users = CourseInstructorRole(self.course.id).users_with_role().values_list('id', flat=True)
+            user_ids.extend(admin_users)
+        else:
+            user_ids = CourseEnrollment.objects.filter(
+                course__id=course_key_str, is_active=True
+            ).values_list('user__id', flat=True)
+
+        unique_user_ids = list(set(user_ids))
+        if self.creator.id in unique_user_ids:
+            unique_user_ids.remove(self.creator.id)
+        return unique_user_ids
+
+    def send_new_thread_created_notification(self):
+        """
+        Send notification based on notification_type
+        """
+        thread_type = self.thread.attributes['thread_type']
+        notification_type = (
+            "new_question_post"
+            if thread_type == "question"
+            else ("new_discussion_post" if thread_type == "discussion" else "")
+        )
+        if notification_type not in ['new_discussion_post', 'new_question_post']:
+            raise ValueError(f'Invalid notification type {notification_type}')
+
+        user_ids = self._create_cohort_course_audience()
+        context = {
+            'username': self.creator.username,
+            'post_title': self.thread.title
+        }
+        self._send_notification(user_ids, notification_type, context)
+
+
+def is_discussion_cohorted(course_key_str):
+    """
+    Returns if the discussion is divided by cohorts
+    """
+    cohort_settings = CourseCohortsSettings.objects.get(course_id=course_key_str)
+    discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
+    return cohort_settings.is_cohorted and discussion_settings.always_divide_inline_discussions
diff --git a/lms/djangoapps/discussion/rest_api/tasks.py b/lms/djangoapps/discussion/rest_api/tasks.py
index 85de47286f9..f5bb20c3d85 100644
--- a/lms/djangoapps/discussion/rest_api/tasks.py
+++ b/lms/djangoapps/discussion/rest_api/tasks.py
@@ -8,7 +8,7 @@
 from lms.djangoapps.courseware.courses import get_course_with_access
 from openedx.core.djangoapps.django_comment_common.comment_client.thread import Thread
 from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
-from lms.djangoapps.discussion.rest_api.utils import DiscussionNotificationSender
+from lms.djangoapps.discussion.rest_api.discussions_notifications import DiscussionNotificationSender
 
 
 User = get_user_model()
@@ -46,3 +46,4 @@ def send_response_notifications(thread_id, course_key_str, user_id, parent_id=No
     notification_sender.send_new_comment_notification()
     notification_sender.send_new_response_notification()
     notification_sender.send_new_comment_on_response_notification()
+    notification_sender.send_response_on_followed_post_notification()
diff --git a/lms/djangoapps/discussion/rest_api/tests/test_api.py b/lms/djangoapps/discussion/rest_api/tests/test_api.py
index 34e323c723e..1193384bfbd 100644
--- a/lms/djangoapps/discussion/rest_api/tests/test_api.py
+++ b/lms/djangoapps/discussion/rest_api/tests/test_api.py
@@ -2156,6 +2156,7 @@ def test_invalid_field(self):
 @disable_signal(api, 'comment_created')
 @disable_signal(api, 'comment_voted')
 @mock.patch.dict("django.conf.settings.FEATURES", {"ENABLE_DISCUSSION_SERVICE": True})
+@mock.patch("lms.djangoapps.discussion.signals.handlers.send_response_notifications", new=mock.Mock())
 class CreateCommentTest(
         ForumsEnableMixin,
         CommentsServiceMockMixin,
@@ -2194,6 +2195,17 @@ def setUp(self):
             "raw_body": "Test body",
         }
 
+        mock_response = {
+            'collection': [],
+            'page': 1,
+            'num_pages': 1,
+            'subscriptions_count': 1,
+            'corrected_text': None
+
+        }
+        self.register_get_subscriptions('cohort_thread', mock_response)
+        self.register_get_subscriptions('test_thread', mock_response)
+
     @ddt.data(None, "test_parent")
     @mock.patch("eventtracking.tracker.emit")
     def test_success(self, parent_id, mock_emit):
diff --git a/lms/djangoapps/discussion/rest_api/tests/test_tasks.py b/lms/djangoapps/discussion/rest_api/tests/test_tasks.py
index e7bc03c0adc..896e237f5f1 100644
--- a/lms/djangoapps/discussion/rest_api/tests/test_tasks.py
+++ b/lms/djangoapps/discussion/rest_api/tests/test_tasks.py
@@ -2,11 +2,14 @@
 Test cases for tasks.py
 """
 from unittest import mock
-from django.conf import settings
-from edx_toggles.toggles.testutils import override_waffle_flag
+from unittest.mock import Mock
 
 import ddt
 import httpretty
+from django.conf import settings
+from edx_toggles.toggles.testutils import override_waffle_flag
+from openedx_events.learning.signals import USER_NOTIFICATION_REQUESTED
+
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.tests.factories import StaffFactory, UserFactory
 from lms.djangoapps.discussion.django_comment_client.tests.factories import RoleFactory
@@ -14,18 +17,19 @@
 from lms.djangoapps.discussion.rest_api.tests.utils import ThreadMock, make_minimal_cs_thread
 from openedx.core.djangoapps.course_groups.models import CohortMembership, CourseCohortsSettings
 from openedx.core.djangoapps.course_groups.tests.helpers import CohortFactory
+from openedx.core.djangoapps.discussions.models import DiscussionTopicLink
 from openedx.core.djangoapps.django_comment_common.models import (
-    CourseDiscussionSettings,
     FORUM_ROLE_COMMUNITY_TA,
     FORUM_ROLE_GROUP_MODERATOR,
     FORUM_ROLE_MODERATOR,
     FORUM_ROLE_STUDENT,
+    CourseDiscussionSettings
 )
-from openedx.core.djangoapps.discussions.models import DiscussionTopicLink
 from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
-from openedx_events.learning.signals import USER_NOTIFICATION_REQUESTED
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
+
+from ..discussions_notifications import DiscussionNotificationSender
 from .test_views import DiscussionAPIViewTestMixin
 
 
@@ -241,6 +245,7 @@ def test_notification_is_send_to_cohort_ids(self, cohort_text, notification_type
         self.assert_users_id_list(user_ids_list, handler.call_args[1]['notification_data'].user_ids)
 
 
+@ddt.ddt
 @override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
 class TestSendResponseNotifications(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
     """
@@ -272,6 +277,7 @@ def setUp(self):
                 "thread_type": 'discussion',
                 "title": thread.title,
             })
+        self._register_subscriptions_endpoint()
 
     def test_basic(self):
         """
@@ -293,8 +299,8 @@ def test_send_notification_to_thread_creator(self):
         # Post the form or do what it takes to send the signal
 
         send_response_notifications(self.thread.id, str(self.course.id), self.user_2.id, parent_id=None)
-        self.assertEqual(handler.call_count, 1)
-        args = handler.call_args[1]['notification_data']
+        self.assertEqual(handler.call_count, 2)
+        args = handler.call_args_list[0][1]['notification_data']
         self.assertEqual([int(user_id) for user_id in args.user_ids], [self.user_1.id])
         self.assertEqual(args.notification_type, 'new_response')
         expected_context = {
@@ -363,13 +369,13 @@ def test_send_notification_to_parent_threads(self):
 
     def test_no_signal_on_creators_own_thread(self):
         """
-        Makes sure that no signal is emitted if user creates response on
+        Makes sure that 1 signal is emitted if user creates response on
         their own thread.
         """
         handler = mock.Mock()
         USER_NOTIFICATION_REQUESTED.connect(handler)
         send_response_notifications(self.thread.id, str(self.course.id), self.user_1.id, parent_id=None)
-        self.assertEqual(handler.call_count, 0)
+        self.assertEqual(handler.call_count, 1)
 
     def test_comment_creators_own_response(self):
         """
@@ -387,7 +393,7 @@ def test_comment_creators_own_response(self):
 
         send_response_notifications(self.thread.id, str(self.course.id), self.user_3.id, parent_id=self.thread_2.id)
         # check if 1 call is made to the handler i.e. for the thread creator
-        self.assertEqual(handler.call_count, 1)
+        self.assertEqual(handler.call_count, 2)
 
         # check if the notification is sent to the thread creator
         args_comment = handler.call_args_list[0][1]['notification_data']
@@ -406,6 +412,71 @@ def test_comment_creators_own_response(self):
         )
         self.assertEqual(args_comment.app_name, 'discussion')
 
+    @ddt.data(
+        (None, 'response_on_followed_post'), (1, 'comment_on_followed_post')
+    )
+    @ddt.unpack
+    def test_send_notification_to_followers(self, parent_id, notification_type):
+        """
+        Test that the notification is sent to the followers of the thread
+        """
+        self.register_get_comment_response({
+            'id': self.thread.id,
+            'thread_id': self.thread.id,
+            'user_id': self.thread.user_id
+        })
+        handler = Mock()
+        USER_NOTIFICATION_REQUESTED.connect(handler)
+
+        # Post the form or do what it takes to send the signal
+        notification_sender = DiscussionNotificationSender(self.thread, self.course, self.user_2, parent_id=parent_id)
+        notification_sender.send_response_on_followed_post_notification()
+        self.assertEqual(handler.call_count, 1)
+        args = handler.call_args[1]['notification_data']
+        # only sent to user_3 because user_2 is the one who created the response
+        self.assertEqual([self.user_3.id], args.user_ids)
+        self.assertEqual(args.notification_type, notification_type)
+        expected_context = {
+            'replier_name': self.user_2.username,
+            'post_title': 'test thread',
+            'course_name': self.course.display_name,
+        }
+        if parent_id:
+            expected_context['author_name'] = 'dummy'
+        self.assertDictEqual(args.context, expected_context)
+        self.assertEqual(
+            args.content_url,
+            _get_mfe_url(self.course.id, self.thread.id)
+        )
+        self.assertEqual(args.app_name, 'discussion')
+
+    def _register_subscriptions_endpoint(self):
+        """
+        Registers the endpoint for the subscriptions API
+        """
+        mock_response = {
+            'collection': [
+                {
+                    '_id': 1,
+                    'subscriber_id': str(self.user_2.id),
+                    "source_id": self.thread.id,
+                    "source_type": "thread",
+                },
+                {
+                    '_id': 2,
+                    'subscriber_id': str(self.user_3.id),
+                    "source_id": self.thread.id,
+                    "source_type": "thread",
+                },
+            ],
+            'page': 1,
+            'num_pages': 1,
+            'subscriptions_count': 2,
+            'corrected_text': None
+
+        }
+        self.register_get_subscriptions(self.thread.id, mock_response)
+
 
 @override_waffle_flag(ENABLE_NOTIFICATIONS, active=True)
 class TestSendCommentNotification(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
@@ -456,6 +527,7 @@ def test_new_comment_notification(self):
             'thread_id': thread.id,
             'user_id': response.user_id
         })
+        self.register_get_subscriptions(1, {})
         send_response_notifications(thread.id, str(self.course.id), self.user_2.id, parent_id=response.id)
         handler.assert_called_once()
         context = handler.call_args[1]['notification_data'].context
diff --git a/lms/djangoapps/discussion/rest_api/tests/test_utils.py b/lms/djangoapps/discussion/rest_api/tests/test_utils.py
index a0e7790cba7..db24847a82f 100644
--- a/lms/djangoapps/discussion/rest_api/tests/test_utils.py
+++ b/lms/djangoapps/discussion/rest_api/tests/test_utils.py
@@ -2,29 +2,29 @@
 Tests for Discussion REST API utils.
 """
 
+import unittest
 from datetime import datetime, timedelta
 
 import ddt
 from pytz import UTC
-import unittest
-from common.djangoapps.student.roles import CourseStaffRole, CourseInstructorRole
-from lms.djangoapps.discussion.django_comment_client.tests.utils import ForumsEnableMixin
-from lms.djangoapps.discussion.rest_api.tests.utils import CommentsServiceMockMixin
-from openedx.core.djangoapps.discussions.models import PostingRestriction, DiscussionsConfiguration
-from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
-from xmodule.modulestore.tests.factories import CourseFactory
 
+from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
 from common.djangoapps.student.tests.factories import CourseEnrollmentFactory, UserFactory
 from lms.djangoapps.discussion.django_comment_client.tests.factories import RoleFactory
+from lms.djangoapps.discussion.django_comment_client.tests.utils import ForumsEnableMixin
+from lms.djangoapps.discussion.rest_api.tests.utils import CommentsServiceMockMixin
 from lms.djangoapps.discussion.rest_api.utils import (
     discussion_open_for_user,
-    get_course_ta_users_list,
+    get_archived_topics,
     get_course_staff_users_list,
+    get_course_ta_users_list,
     get_moderator_users_list,
-    get_archived_topics,
-    remove_empty_sequentials,
-    is_posting_allowed
+    is_posting_allowed,
+    remove_empty_sequentials
 )
+from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration, PostingRestriction
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import CourseFactory
 
 
 class DiscussionAPIUtilsTestCase(ModuleStoreTestCase):
diff --git a/lms/djangoapps/discussion/rest_api/tests/test_views.py b/lms/djangoapps/discussion/rest_api/tests/test_views.py
index 0e303cc2409..6ac64f161a9 100644
--- a/lms/djangoapps/discussion/rest_api/tests/test_views.py
+++ b/lms/djangoapps/discussion/rest_api/tests/test_views.py
@@ -2357,6 +2357,7 @@ def test_delete_nonexistent_comment(self):
 @httpretty.activate
 @disable_signal(api, 'comment_created')
 @mock.patch.dict("django.conf.settings.FEATURES", {"ENABLE_DISCUSSION_SERVICE": True})
+@mock.patch("lms.djangoapps.discussion.signals.handlers.send_response_notifications", new=mock.Mock())
 class CommentViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
     """Tests for CommentViewSet create"""
     def setUp(self):
diff --git a/lms/djangoapps/discussion/rest_api/tests/utils.py b/lms/djangoapps/discussion/rest_api/tests/utils.py
index 5db131bec88..39dddcf33f3 100644
--- a/lms/djangoapps/discussion/rest_api/tests/utils.py
+++ b/lms/djangoapps/discussion/rest_api/tests/utils.py
@@ -425,6 +425,18 @@ def register_user_active_threads(self, user_id, response):
             status=200
         )
 
+    def register_get_subscriptions(self, thread_id, response):
+        """
+        Register a mock response for GET on the CS comment active threads endpoint
+        """
+        assert httpretty.is_enabled(), 'httpretty must be enabled to mock calls.'
+        httpretty.register_uri(
+            httpretty.GET,
+            f"http://localhost:4567/api/v1/threads/{thread_id}/subscriptions",
+            body=json.dumps(response),
+            status=200
+        )
+
     def assert_query_params_equal(self, httpretty_request, expected_params):
         """
         Assert that the given mock request had the expected query parameters
diff --git a/lms/djangoapps/discussion/rest_api/utils.py b/lms/djangoapps/discussion/rest_api/utils.py
index 0b84cf2ac63..e7dca499100 100644
--- a/lms/djangoapps/discussion/rest_api/utils.py
+++ b/lms/djangoapps/discussion/rest_api/utils.py
@@ -2,33 +2,23 @@
 Utils for discussion API.
 """
 from datetime import datetime
-from pytz import UTC
-from typing import List, Dict
+from typing import Dict, List
 
-from django.conf import settings
 from django.contrib.auth.models import User  # lint-amnesty, pylint: disable=imported-auth-user
 from django.core.paginator import Paginator
 from django.db.models.functions import Length
-from django.utils.translation import gettext_lazy as _
+from pytz import UTC
 
-from common.djangoapps.student.models import CourseEnrollment
-from common.djangoapps.student.roles import CourseStaffRole, CourseInstructorRole
-from lms.djangoapps.discussion.django_comment_client.permissions import get_team
+from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
 from lms.djangoapps.discussion.django_comment_client.utils import has_discussion_privileges
-from openedx.core.djangoapps.course_groups.models import CourseCohortsSettings, CourseUserGroup
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration, PostingRestriction
-from openedx.core.djangoapps.discussions.utils import get_divided_discussions
 from openedx.core.djangoapps.django_comment_common.models import (
-    Role,
-    CourseDiscussionSettings,
     FORUM_ROLE_ADMINISTRATOR,
-    FORUM_ROLE_MODERATOR,
-    FORUM_ROLE_GROUP_MODERATOR,
     FORUM_ROLE_COMMUNITY_TA,
+    FORUM_ROLE_GROUP_MODERATOR,
+    FORUM_ROLE_MODERATOR,
+    Role
 )
-from openedx_events.learning.signals import USER_NOTIFICATION_REQUESTED
-from openedx_events.learning.data import UserNotificationData
-from openedx.core.djangoapps.django_comment_common.comment_client.comment import Comment
 
 
 class AttributeDict(dict):
@@ -371,196 +361,6 @@ def get_archived_topics(filtered_topic_ids: List[str], topics: List[Dict[str, st
     return archived_topics
 
 
-def is_discussion_cohorted(course_key_str):
-    """
-    Returns if the discussion is divided by cohorts
-    """
-    cohort_settings = CourseCohortsSettings.objects.get(course_id=course_key_str)
-    discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
-    return cohort_settings.is_cohorted and discussion_settings.always_divide_inline_discussions
-
-
-class DiscussionNotificationSender:
-    """
-    Class to send notifications to users who are subscribed to the thread.
-    """
-
-    def __init__(self, thread, course, creator, parent_id=None):
-        self.thread = thread
-        self.course = course
-        self.creator = creator
-        self.parent_id = parent_id
-        self.parent_response = None
-        self._get_parent_response()
-
-    def _send_notification(self, user_ids, notification_type, extra_context=None):
-        """
-        Send notification to users
-        """
-        if not user_ids:
-            return
-
-        if extra_context is None:
-            extra_context = {}
-
-        notification_data = UserNotificationData(
-            user_ids=[int(user_id) for user_id in user_ids],
-            context={
-                "replier_name": self.creator.username,
-                "post_title": self.thread.title,
-                "course_name": self.course.display_name,
-                **extra_context,
-            },
-            notification_type=notification_type,
-            content_url=f"{settings.DISCUSSIONS_MICROFRONTEND_URL}/{str(self.course.id)}/posts/{self.thread.id}",
-            app_name="discussion",
-            course_key=self.course.id,
-        )
-        USER_NOTIFICATION_REQUESTED.send_event(notification_data=notification_data)
-
-    def _get_parent_response(self):
-        """
-        Get parent response object
-        """
-        if self.parent_id and not self.parent_response:
-            self.parent_response = Comment(id=self.parent_id).retrieve()
-
-        return self.parent_response
-
-    def _create_cohort_course_audience(self):
-        """
-        Creates audience based on user cohort and role
-        """
-        course_key_str = str(self.course.id)
-        discussion_cohorted = is_discussion_cohorted(course_key_str)
-
-        # Retrieves cohort divided discussion
-        discussion_settings = CourseDiscussionSettings.objects.get(course_id=course_key_str)
-        divided_course_wide_discussions, divided_inline_discussions = get_divided_discussions(
-            self.course,
-            discussion_settings
-        )
-
-        # Checks if post has any cohort assigned
-        group_id = self.thread.attributes['group_id']
-        if group_id is not None:
-            group_id = int(group_id)
-
-        # Course wide topics
-        topic_id = self.thread.attributes['commentable_id']
-        all_topics = divided_inline_discussions + divided_course_wide_discussions
-        topic_divided = topic_id in all_topics or discussion_settings.always_divide_inline_discussions
-
-        # Team object from topic id
-        team = get_team(topic_id)
-
-        user_ids = []
-        if team:
-            user_ids = team.users.all().values_list('id', flat=True)
-        elif discussion_cohorted and topic_divided and group_id is not None:
-            users_in_cohort = CourseUserGroup.objects.filter(
-                course_id=course_key_str, id=group_id
-            ).values_list('users__id', flat=True)
-            user_ids.extend(users_in_cohort)
-
-            privileged_roles = [FORUM_ROLE_ADMINISTRATOR, FORUM_ROLE_MODERATOR, FORUM_ROLE_COMMUNITY_TA]
-            privileged_users = Role.objects.filter(
-                name__in=privileged_roles,
-                course_id=course_key_str
-            ).values_list('users__id', flat=True)
-            user_ids.extend(privileged_users)
-
-            staff_users = CourseStaffRole(self.course.id).users_with_role().values_list('id', flat=True)
-            user_ids.extend(staff_users)
-
-            admin_users = CourseInstructorRole(self.course.id).users_with_role().values_list('id', flat=True)
-            user_ids.extend(admin_users)
-        else:
-            user_ids = CourseEnrollment.objects.filter(
-                course__id=course_key_str, is_active=True
-            ).values_list('user__id', flat=True)
-
-        unique_user_ids = list(set(user_ids))
-        if self.creator.id in unique_user_ids:
-            unique_user_ids.remove(self.creator.id)
-        return unique_user_ids
-
-    def send_new_response_notification(self):
-        """
-        Send notification to users who are subscribed to the main thread/post i.e.
-        there is a response to the main thread.
-        """
-        if not self.parent_id and self.creator.id != int(self.thread.user_id):
-            self._send_notification([self.thread.user_id], "new_response")
-
-    def _response_and_thread_has_same_creator(self) -> bool:
-        """
-        Check if response and main thread have same author.
-        """
-        return int(self.parent_response.user_id) == int(self.thread.user_id)
-
-    def _response_and_comment_has_same_creator(self):
-        return int(self.parent_response.attributes['user_id']) == self.creator.id
-
-    def send_new_comment_notification(self):
-        """
-        Send notification to parent thread creator i.e. comment on the response.
-        """
-        if (
-            self.parent_response and
-            self.creator.id != int(self.thread.user_id)
-        ):
-            # use your if author of response is same as author of post.
-            # use 'their' if comment author is also response author.
-            author_name = (
-                # Translators: Replier commented on "your" response to your post
-                _("your")
-                if self._response_and_thread_has_same_creator()
-                else (
-                    # Translators: Replier commented on "their" response to your post
-                    _("their")
-                    if self._response_and_comment_has_same_creator()
-                    else f"{self.parent_response.username}'s"
-                )
-            )
-            context = {
-                "author_name": str(author_name),
-            }
-            self._send_notification([self.thread.user_id], "new_comment", extra_context=context)
-
-    def send_new_comment_on_response_notification(self):
-        """
-        Send notification to parent response creator i.e. comment on the response.
-        Do not send notification if author of response is same as author of post.
-        """
-        if (
-            self.parent_response and
-            self.creator.id != int(self.parent_response.user_id) and not
-            self._response_and_thread_has_same_creator()
-        ):
-            self._send_notification([self.parent_response.user_id], "new_comment_on_response")
-
-    def send_new_thread_created_notification(self):
-        """
-        Send notification based on notification_type
-        """
-        thread_type = self.thread.attributes['thread_type']
-        notification_type = (
-            "new_question_post"
-            if thread_type == "question"
-            else ("new_discussion_post" if thread_type == "discussion" else "")
-        )
-        if notification_type not in ['new_discussion_post', 'new_question_post']:
-            raise ValueError(f'Invalid notification type {notification_type}')
-
-        user_ids = self._create_cohort_course_audience()
-        context = {
-            'username': self.creator.username,
-            'post_title': self.thread.title
-        }
-        self._send_notification(user_ids, notification_type, context)
-
-
 def is_posting_allowed(posting_restrictions: str, blackout_schedules: List):
     """
     Check if posting is allowed based on the given posting restrictions and blackout schedules.
diff --git a/openedx/core/djangoapps/django_comment_common/comment_client/subscriptions.py b/openedx/core/djangoapps/django_comment_common/comment_client/subscriptions.py
new file mode 100644
index 00000000000..545948a092c
--- /dev/null
+++ b/openedx/core/djangoapps/django_comment_common/comment_client/subscriptions.py
@@ -0,0 +1,50 @@
+"""
+Subscription model is used to get users who are subscribed to the main thread/post i.e.
+"""
+import logging
+
+from . import models, settings, utils
+
+log = logging.getLogger(__name__)
+
+
+class Subscription(models.Model):
+    """
+    Subscription model is used to get users who are subscribed to the main thread/post i.e.
+    """
+    # accessible_fields can be set and retrieved on the model
+    accessible_fields = [
+        '_id', 'subscriber_id', "source_id", "source_type"
+    ]
+
+    type = 'subscriber'
+    base_url = f"{settings.PREFIX}/threads"
+
+    @classmethod
+    def fetch(cls, thread_id, query_params):
+        """
+        Fetches the subscriptions for a given thread_id
+        """
+        params = {
+            'page': query_params.get('page', 1),
+            'per_page': query_params.get('per_page', 20),
+            'id': thread_id
+        }
+        params.update(
+            utils.strip_blank(utils.strip_none(query_params))
+        )
+        response = utils.perform_request(
+            'get',
+            cls.url(action='get', params=params) + "/subscriptions",
+            params,
+            metric_tags=[],
+            metric_action='subscription.get',
+            paged_results=True
+        )
+        return utils.SubscriptionsPaginatedResult(
+            collection=response.get('collection', []),
+            page=response.get('page', 1),
+            num_pages=response.get('num_pages', 1),
+            subscriptions_count=response.get('subscriptions_count', 0),
+            corrected_text=response.get('corrected_text', None)
+        )
diff --git a/openedx/core/djangoapps/django_comment_common/comment_client/thread.py b/openedx/core/djangoapps/django_comment_common/comment_client/thread.py
index 74e3e0eba27..6f92313b6ee 100644
--- a/openedx/core/djangoapps/django_comment_common/comment_client/thread.py
+++ b/openedx/core/djangoapps/django_comment_common/comment_client/thread.py
@@ -11,7 +11,6 @@
 
 
 class Thread(models.Model):
-
     # accessible_fields can be set and retrieved on the model
     accessible_fields = [
         'id', 'title', 'body', 'anonymous', 'anonymous_to_peers', 'course_id',
diff --git a/openedx/core/djangoapps/django_comment_common/comment_client/utils.py b/openedx/core/djangoapps/django_comment_common/comment_client/utils.py
index 2432ab36be8..a67cdbdbc48 100644
--- a/openedx/core/djangoapps/django_comment_common/comment_client/utils.py
+++ b/openedx/core/djangoapps/django_comment_common/comment_client/utils.py
@@ -131,6 +131,17 @@ def __init__(self, collection, page, num_pages, thread_count=0, corrected_text=N
         self.corrected_text = corrected_text
 
 
+class SubscriptionsPaginatedResult:
+    """ class for paginated results returned from comment services"""
+
+    def __init__(self, collection, page, num_pages, subscriptions_count=0, corrected_text=None):
+        self.collection = collection
+        self.page = page
+        self.num_pages = num_pages
+        self.subscriptions_count = subscriptions_count
+        self.corrected_text = corrected_text
+
+
 def check_forum_heartbeat():
     """
     Check the forum connection via its built-in heartbeat service and create an answer which can be used in the LMS
diff --git a/openedx/core/djangoapps/notifications/base_notification.py b/openedx/core/djangoapps/notifications/base_notification.py
index d7d5f58574b..576283eadc1 100644
--- a/openedx/core/djangoapps/notifications/base_notification.py
+++ b/openedx/core/djangoapps/notifications/base_notification.py
@@ -77,7 +77,42 @@
             'username': 'Post author name',
         },
         'email_template': '',
-    }
+    },
+    'response_on_followed_post': {
+        'notification_app': 'discussion',
+        'name': 'response_on_followed_post',
+        'is_core': True,
+        'web': False,
+        'email': False,
+        'push': False,
+        'info': '',
+        'non_editable': [],
+        'content_template': _('<{p}><{strong}>{replier_name}</{strong}> responded to a post you’re following: '
+                              '<{strong}>{post_title}</{strong}></{p}>'),
+        'content_context': {
+            'post_title': 'Post title',
+            'replier_name': 'replier name',
+        },
+        'email_template': '',
+    },
+    'comment_on_followed_post': {
+        'notification_app': 'discussion',
+        'name': 'comment_on_followed_post',
+        'is_core': True,
+        'web': False,
+        'email': False,
+        'push': False,
+        'info': '',
+        'non_editable': [],
+        'content_template': _('<{p}><{strong}>{replier_name}</{strong}> commented on {author_name}\'s response in '
+                              'a post you’re following <{strong}>{post_title}</{strong}></{p}>'),
+        'content_context': {
+            'post_title': 'Post title',
+            'author_name': 'author name',
+            'replier_name': 'replier name',
+        },
+        'email_template': '',
+    },
 }
 
 COURSE_NOTIFICATION_APPS = {
diff --git a/openedx/core/djangoapps/notifications/tests/test_views.py b/openedx/core/djangoapps/notifications/tests/test_views.py
index 88fd9619ef4..677d72fc84c 100644
--- a/openedx/core/djangoapps/notifications/tests/test_views.py
+++ b/openedx/core/djangoapps/notifications/tests/test_views.py
@@ -217,7 +217,13 @@ def _expected_api_response(self):
             'notification_preference_config': {
                 'discussion': {
                     'enabled': True,
-                    'core_notification_types': ['new_comment_on_response', 'new_comment', 'new_response'],
+                    'core_notification_types': [
+                        'new_comment_on_response',
+                        'new_comment',
+                        'new_response',
+                        'response_on_followed_post',
+                        'comment_on_followed_post'
+                    ],
                     'notification_types': {
                         'core': {
                             'web': True,
@@ -227,7 +233,7 @@ def _expected_api_response(self):
                                     'following, including endorsements to your responses and on your posts.'
                         },
                         'new_discussion_post': {'web': False, 'email': False, 'push': False, 'info': ''},
-                        'new_question_post': {'web': False, 'email': False, 'push': False, 'info': ''}
+                        'new_question_post': {'web': False, 'email': False, 'push': False, 'info': ''},
                     },
                     'non_editable': {
                         'core': ['web']