server: Exit consume-redis-stream loop on error (#5948)

* server: Exit consume-redis-stream loop on error

Recently a yield was added to the consume-redis-stream loop. It is
possible that when the fiber is scheduled again, things have changed
(likely because the redis acks fiber caused a shutdown of the socket).
If so we will exit from the loop instead of trying to read from a
possibly closed socket.

Signed-off-by: Abhijat Malviya <[email protected]>

Author: abhijat

src/server/replica.cc

@@ -704,6 +704,17 @@ error_code Replica::ConsumeRedisStream() {
       ThisFiber::Yield();
     }
 
+    // If the acks-fb or something else triggered a shutdown, then do not attempt to read from the
+    // stream.
+    if (!exec_st_.IsRunning()) {
+      DCHECK(exec_st_.IsError());
+      LOG_REPL_ERROR("Stopping stream consumer in phase "
+                     << GetCurrentPhase()
+                     << " because of external error: " << exec_st_.GetError().Format());
+      acks_fb_.JoinIfNeeded();
+      return exec_st_.GetError();
+    }
+
     auto response = ReadRespReply(&io_buf, /*copy_msg=*/false);
     if (!response.has_value()) {
       LOG_REPL_ERROR("Error in Redis Stream at phase "