diff options
| author | Arnaud Cogoluègnes <acogoluegnes@gmail.com> | 2022-02-07 10:09:09 +0100 |
|---|---|---|
| committer | Arnaud Cogoluègnes <acogoluegnes@gmail.com> | 2022-05-09 10:52:35 +0200 |
| commit | adba97fe6ecf02fcb5ef3fa698045162fc3e9d31 (patch) | |
| tree | 9fe73848d2ea0bd6a04d8132d453cbd0755cb984 | |
| parent | 827fe6f008afa0eb4b199ec34762a2bed4310b8c (diff) | |
| download | rabbitmq-server-git-adba97fe6ecf02fcb5ef3fa698045162fc3e9d31.tar.gz | |
Do not dispatch on credit request is consumer is inactive
A formerly active consumer can have in-flight credit
requests when it becomes inactive. This commit checks
the state of consumer on credit requests and make sure
not to dispatch messages if it's inactive.
| -rw-r--r-- | deps/rabbitmq_stream/src/rabbit_stream_reader.erl | 18 |
1 files changed, 15 insertions, 3 deletions
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl index bce0864b07..bda0f35dab 100644 --- a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl +++ b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl @@ -1983,9 +1983,22 @@ handle_frame_post_auth(Transport, Connection, #stream_connection_state{consumers = Consumers} = State, {credit, SubscriptionId, Credit}) -> - %% FIXME check consumer is active - %% if not active, don't dispatch, just return appropriate response case Consumers of + #{SubscriptionId := #consumer{log = undefined}} -> + %% the consumer is not active, it's likely to be credit leftovers + %% from a formerly active consumer, just logging and send an error + rabbit_log:debug("Giving credit to an inactive consumer: ~p", + [SubscriptionId]), + + Code = ?RESPONSE_CODE_PRECONDITION_FAILED, + Frame = + rabbit_stream_core:frame({response, 1, + {credit, Code, SubscriptionId}}), + send(Transport, S, Frame), + rabbit_global_counters:increase_protocol_counter(stream, + ?PRECONDITION_FAILED, + 1), + {Connection, State}; #{SubscriptionId := Consumer} -> #consumer{credit = AvailableCredit, last_listener_offset = LLO} = Consumer, @@ -3294,7 +3307,6 @@ consumer_i(activity_status, consumer_i(_Unknown, _) -> ?UNKNOWN_FIELD. - publishers_info(Pid, InfoItems) -> gen_server2:call(Pid, {publishers_info, InfoItems}). |