Implement `is_unresponsive` for stream queuesis-unresponsive-stream

1 files changed, 11 insertions, 0 deletions

diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
index dccc725a9d..3704cf6ecc 100644
--- a/deps/rabbit/src/rabbit_amqqueue.erl
+++ b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -1207,6 +1207,17 @@ is_unresponsive(Q, Timeout) when ?amqqueue_is_quorum(Q) ->
     catch
         exit:{timeout, _} ->
             true
+    end;
+is_unresponsive(Q, Timeout) when ?amqqueue_is_stream(Q) ->
+    try
+        #{leader_pid := LeaderPid} = amqqueue:get_type_state(Q),
+        case gen_batch_server:call(LeaderPid, get_reader_context, Timeout) of
+            #{dir := _} -> false;
+            _ -> true
+        end
+    catch
+        exit:{timeout, _} ->
+            true
     end.
 
 format(Q) when ?amqqueue_is_quorum(Q) -> rabbit_quorum_queue:format(Q);