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

fix(bufworker): do not avoid retry if inflight table is full 

Otherwise there's no other piece of code that would retry the inflight
queries in that case.
This commit is contained in:
Andrew Mayorov 2023-02-08 14:03:40 +03:00
parent d8d06a260f
commit c6fc0ec8cd
No known key found for this signature in database
GPG Key ID: 2837C62ACFBFED5D
1 changed files with 2 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
apps/emqx_resource/src/emqx_resource_buffer_worker.erl
View File

@ -334,12 +334,7 @@ resume_from_blocked(Data) ->
{single, Ref, Query} -> {single, Ref, Query} ->
%% We retry msgs in inflight window sync, as if we send them %% We retry msgs in inflight window sync, as if we send them
%% async, they will be appended to the end of inflight window again. %% async, they will be appended to the end of inflight window again.
case is_inflight_full(InflightTID) of retry_inflight_sync(Ref, Query, Data);
true ->
{keep_state, Data};
false ->
retry_inflight_sync(Ref, Query, Data)
end;
{batch, Ref, NotExpired, Expired} -> {batch, Ref, NotExpired, Expired} ->
update_inflight_item(InflightTID, Ref, NotExpired), update_inflight_item(InflightTID, Ref, NotExpired),
NumExpired = length(Expired), NumExpired = length(Expired),
@ -347,12 +342,7 @@ resume_from_blocked(Data) ->
NumExpired > 0 andalso ?tp(buffer_worker_retry_expired, #{expired => Expired}), NumExpired > 0 andalso ?tp(buffer_worker_retry_expired, #{expired => Expired}),
%% We retry msgs in inflight window sync, as if we send them %% We retry msgs in inflight window sync, as if we send them
%% async, they will be appended to the end of inflight window again. %% async, they will be appended to the end of inflight window again.
case is_inflight_full(InflightTID) of retry_inflight_sync(Ref, NotExpired, Data)
true ->
{keep_state, Data};
false ->
retry_inflight_sync(Ref, NotExpired, Data)
end
end. end.
retry_inflight_sync(Ref, QueryOrBatch, Data0) -> retry_inflight_sync(Ref, QueryOrBatch, Data0) ->