diff --git a/internal/reader/reader.go b/internal/reader/reader.go
index 5c44d3b..79ec8e9 100644
--- a/internal/reader/reader.go
+++ b/internal/reader/reader.go
@@ -216,7 +216,11 @@ func ReadTopic(ctx context.Context, streamInfo StreamInfo, reader *client.TopicR
 	for ctx.Err() == nil {
 		msg, err := reader.ReadMessage(ctx)
 		if err != nil {
-			xlog.Fatal(ctx, "Unable to read message", zap.Error(err))
+			if ctx.Err() != nil {
+				xlog.Error(ctx, "Unable to read message", zap.Error(err))
+			} else {
+				xlog.Fatal(ctx, "Unable to read message, fatal error ctx is not cancelled", zap.Error(err))
+			}
 			return
 		}