yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

wip(sessds): handle recoverable errors in stream scheduler

This commit is contained in:
Andrew Mayorov 2024-03-26 18:23:08 +01:00
parent c6085a6ab0
commit 393a1b1391
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
2 changed files with 19 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx/src/emqx_persistent_session_ds.erl
View File

@ -569,7 +569,7 @@ replay_streams(Session0 = #{replay := [{_StreamKey, Srs0} | Rest]}, ClientInfo)
RetryTimeout = ?TIMEOUT_RETRY_REPLAY,
?SLOG(warning, #{
msg => "failed_to_fetch_replay_batch",
stream => Srs0,
iterator => Srs0#srs.it_end,
reason => Reason,
class => recoverable,
retry_in_ms => RetryTimeout

14
apps/emqx/src/emqx_persistent_session_ds_stream_scheduler.erl
View File

@ -208,9 +208,8 @@ ensure_iterator(TopicFilter, StartTime, SubId, {{RankX, RankY}, Stream}, S) ->
?SLOG(debug, #{
msg => new_stream, key => Key, stream => Stream
}),
{ok, Iterator} = emqx_ds:make_iterator(
?PERSISTENT_MESSAGE_DB, Stream, TopicFilter, StartTime
),
case emqx_ds:make_iterator(?PERSISTENT_MESSAGE_DB, Stream, TopicFilter, StartTime) of
{ok, Iterator} ->
NewStreamState = #srs{
rank_x = RankX,
rank_y = RankY,
@ -218,6 +217,15 @@ ensure_iterator(TopicFilter, StartTime, SubId, {{RankX, RankY}, Stream}, S) ->
it_end = Iterator
},
emqx_persistent_session_ds_state:put_stream(Key, NewStreamState, S);
{error, recoverable, Reason} ->
?SLOG(warning, #{
msg => "failed_to_initialize_stream_iterator",
stream => {Key, Stream},
class => recoverable,
reason => Reason
}),
S
end;
#srs{} ->
S
end.