diff --git a/zerver/lib/actions.py b/zerver/lib/actions.py
index c3ec579444..d7a6e3ebfb 100644
--- a/zerver/lib/actions.py
+++ b/zerver/lib/actions.py
@@ -2777,6 +2777,7 @@ def bulk_remove_subscriptions(users: Iterable[UserProfile],
         notify_subscriptions_removed(user_profile, streams_by_user[user_profile.id])
 
         event = {'type': 'mark_stream_messages_as_read',
+                 'client_id': acting_client.id,
                  'user_profile_id': user_profile.id,
                  'stream_ids': [stream.id for stream in streams]}
         queue_json_publish("deferred_work", event)
@@ -3578,7 +3579,8 @@ def do_mark_all_as_read(user_profile: UserProfile, client: Client) -> int:
     return count
 
 def do_mark_stream_messages_as_read(user_profile: UserProfile,
-                                    stream: Optional[Stream],
+                                    client: Client,
+                                    stream: Stream,
                                     topic_name: Optional[str]=None) -> int:
     log_statsd_event('mark_stream_as_read')
 
diff --git a/zerver/views/messages.py b/zerver/views/messages.py
index e1bc0a35d5..06b5cdef7a 100644
--- a/zerver/views/messages.py
+++ b/zerver/views/messages.py
@@ -999,7 +999,7 @@ def mark_stream_as_read(request: HttpRequest,
                         user_profile: UserProfile,
                         stream_id: int=REQ(validator=check_int)) -> HttpResponse:
     stream, recipient, sub = access_stream_by_id(user_profile, stream_id)
-    count = do_mark_stream_messages_as_read(user_profile, stream)
+    count = do_mark_stream_messages_as_read(user_profile, request.client, stream)
 
     log_data_str = "[%s updated]" % (count,)
     request._log_data["extra"] = log_data_str
@@ -1021,7 +1021,7 @@ def mark_topic_as_read(request: HttpRequest,
         if not topic_exists:
             raise JsonableError(_('No such topic \'%s\'') % (topic_name,))
 
-    count = do_mark_stream_messages_as_read(user_profile, stream, topic_name)
+    count = do_mark_stream_messages_as_read(user_profile, request.client, stream, topic_name)
 
     log_data_str = "[%s updated]" % (count,)
     request._log_data["extra"] = log_data_str
diff --git a/zerver/worker/queue_processors.py b/zerver/worker/queue_processors.py
index e1cd534b5e..f145737372 100644
--- a/zerver/worker/queue_processors.py
+++ b/zerver/worker/queue_processors.py
@@ -14,7 +14,8 @@ from django.core.handlers.wsgi import WSGIRequest
 from django.core.handlers.base import BaseHandler
 from zerver.models import \
     get_client, get_system_bot, ScheduledEmail, PreregistrationUser, \
-    get_user_profile_by_id, Message, Realm, Service, UserMessage, UserProfile
+    get_user_profile_by_id, Message, Realm, Service, UserMessage, UserProfile, \
+    Client
 from zerver.lib.context_managers import lockfile
 from zerver.lib.error_notify import do_report_error
 from zerver.lib.feedback import handle_feedback
@@ -41,7 +42,6 @@ from zerver.lib.str_utils import force_str
 from zerver.context_processors import common_context
 from zerver.lib.outgoing_webhook import do_rest_call, get_outgoing_webhook_service_handler
 from zerver.models import get_bot_services
-from zulip import Client
 from zulip_bots.lib import extract_query_without_mention
 from zerver.lib.bot_lib import EmbeddedBotHandler, get_bot_handler, EmbeddedBotQuitException
 
@@ -522,6 +522,7 @@ class DeferredWorker(QueueProcessingWorker):
     def consume(self, event: Mapping[str, Any]) -> None:
         if event['type'] == 'mark_stream_messages_as_read':
             user_profile = get_user_profile_by_id(event['user_profile_id'])
+            client = Client.objects.get(id=event['client_id'])
 
             for stream_id in event['stream_ids']:
                 # Since the user just unsubscribed, we don't require
@@ -529,4 +530,4 @@ class DeferredWorker(QueueProcessingWorker):
                 # streams would never be accessible)
                 (stream, recipient, sub) = access_stream_by_id(user_profile, stream_id,
                                                                require_active=False)
-                do_mark_stream_messages_as_read(user_profile, stream)
+                do_mark_stream_messages_as_read(user_profile, client, stream)