amqp_client direct_consumer not closing channel on process down

[nulian]

I know you need to close channels yourself but the direct consumer says that if the case isn't handled it should close the channel. But the exit message of the direct_consumer causes the channel genserver to go down but it doesn't actually close the channel which causes when you open a new channel the error below.

11:06:31.022 [warn] Connection (#PID<0.289.0>) closing: received hard error {:"connection.close", 504, "CHANNEL_ERROR - second 'channel.open' seen", 20, 10} from server

The example case is written in elixir and using the wrapper amqp. But I also tried the erlang amqp_direct_consumer directly and it gave precisely the same error.

Made this issue here because I think it's a bug in the amqp_client package.

Original elixir package ticket: pma/amqp#186

Repro case is using this package and the below calls.
https://github.com/nulian/rabbit_test

You could replace line 12 in channel_opener.ex with next line for the erlang amqp_client version.
{:ok, channel} = AMQP.Channel.open(conn, {:amqp_direct_consumer, [self()]})

iex -S mix

iex(1)> RabbitSupervisor.start_child()
{:ok, #PID<0.284.0>}
iex(2)> RabbitSupervisor.start_child()
{:ok, #PID<0.290.0>}
iex(3)> RabbitSupervisor.start_child()
{:ok, #PID<0.296.0>}
iex(4)> {_, pid} = v()                                          
{:ok, #PID<0.296.0>}
iex(5)> DynamicSupervisor.terminate_child(RabbitSupervisor, pid)
:ok
iex(6)> 11:07:28.784 [error] gen_server <0.298.0> terminated with reason: {error,{consumer_died,shutdown}}

11:07:28.784 [error] GenServer #PID<0.298.0> terminating
** (stop) {:error, {:consumer_died, :shutdown}}
Last message: {:DOWN, #Reference<0.3377027747.3486515201.88223>, :process, #PID<0.296.0>, :shutdown}
State: {:state, AMQP.DirectConsumer, #PID<0.296.0>}
11:07:28.788 [error] CRASH REPORT Process <0.298.0> with 0 neighbours exited with reason: {error,{consumer_died,shutdown}} in gen_server2:terminate/3 line 1183
11:07:28.788 [error] Supervisor {<0.297.0>,amqp_channel_sup} had child gen_consumer started with amqp_gen_consumer:start_link('Elixir.AMQP.DirectConsumer', <0.296.0>, {<<"client 127.0.0.1:47123 -> 127.0.0.1:5672">>,3}) at <0.298.0> exit with reason {error,{consumer_died,shutdown}} in context child_terminated
11:07:28.788 [error] Supervisor {<0.297.0>,amqp_channel_sup} had child gen_consumer started with amqp_gen_consumer:start_link('Elixir.AMQP.DirectConsumer', <0.296.0>, {<<"client 127.0.0.1:47123 -> 127.0.0.1:5672">>,3}) at <0.298.0> exit with reason reached_max_restart_intensity in context shutdown

[michaelklishin]

I will convert this issue to a GitHub discussion. Currently GitHub will automatically close and lock the issue even though your question will be transferred and responded to elsewhere. This is to let you know that we do not intend to ignore this but this is how the current GitHub conversion mechanism makes it seem for the users :(

[michaelklishin]

@nulian I'm afraid I don't understand what your expectations, in particular

the direct consumer says that if the case isn't handled it should close the channel

amqp_direct_consumer docs say

%% Warning! It is not recommended to rely on a consumer on killing off the
%% channel (through the exit signal). That may cause messages to get lost.
%% Always use amqp_channel:close/{1,3} for a clean shut down.

In general, consumers in RabbitMQ clients do not close channels when they run into an unhandled exception. Java client behaved this
way years ago and most users did not like it (it caught them by surprise), so now there are consumer implementations of both types,
"strict" and "forgiving", and I'm pretty sure the majority of the ecosystem these days uses the latter.

In addition it matters if the consumer was shut down cleanly or terminated abnormally. In case of a clean shutdown
I guess consumers could shut down their channels but I would argue that it's none of their business. The process that
starts both (opens the channel, spawn/starts a consumer) could decide what to do and why.

Erlang (and Elixir by proxy) allows you to link two processes so that they go down together. Libraries such as the RabbitMQ AMQP 0-9-1 client or this Elixir façade to it really should not be too opinionated because usually it is easy enough for your
to express how exactly you want process tree shutdown to happen.

[nulian]

If the channel genserver/supervisors would stay up I wouldn't have had a big problem with the current way it works.
But at the moment when the consumer is taken down with the error the erlang supervisor/genserver that monitors the channel also goes down so the library thinks the channel is no longer being used.
But the channel stays open on the connection without being monitored anymore. Meaning the channel id is still in use while the library thinks it's not.

So the next time you open a new channel you get the error.
{:"connection.close", 504, "CHANNEL_ERROR - second 'channel.open' seen", 20, 10}
And then it takes down the entire connection because the client get's in a unrecoverable state.
That seems to me not working like intended when you can no longer open a new channel on the current connection.

[michaelklishin]

That sounds like a completely separate problem and the solution is not to make consumer processes manage their channels. How can that one be reproduced? (without Elixir and wrappers)

[nulian]

Sorry don't have erlang example because of not knowing erlang very well.

But the direct_consumer can trigger that issue.

So would say the erlang equivalent of the following elixir with the amqp_client erlang modules.

:amqp_connection.open_channel(conn.pid, {:amqp_direct_consumer, [self()]})

Then kill that process of self which triggers the error.
Then on the next open_channel with the same connection pid it will return the second 'channel.open' error

[nulian]

Also saw it before I used direct consumer with channel probably when there was a consumer attached to the channel when I just linked the channel.

So the problem is probably correct what you said that the consumer takes down the channel which then happens in an incorrect way.