On stream restart, the mock stops working

[mdiasribeiro]

I got the following set-up: Producer actor -> Akka stream -> alpakka-amqp -> Consumer akka stream

When I restart either stream (which implies draining the stream, canceling upstream and completing downstream, and then creating a new stream with the same AmqpConnectionProvider and configs), the mock stops working with the new stream.

Using messageCount and consumerCount on the used channel shows the correct number of messages waiting to be read and the new consumer in there (for example, starts at one, goes to 0 when the stream is shutdown and back to 1 when it is recreated), but nothing is delivered to the stream.

Using the same set-up on a real RMQ server instead of the mock seems to be working. Any idea?

[ledoyen]

Hi, thanks for the feedback, It's great to know that this tool is used with an akka stack 😄

I am currently on vacations, and answering from my mobile phone. I will be able to look at this issue in september.

In the meantime, if you want to look further by yourself, my clue would be to look at the two methods close and notifyDeleted in the MockQueue class which interrupt the delivery loop but do not drain past messages and does not block new consumers from being declared.

Edit: I was thinking that it may also be due to the consumer cancelation, which currently does not nack pending messages.

[mdiasribeiro]

Thanks for the response. It might indeed be the delivery loop being interrupted. I enabled debug logging, and I still see messages being published normally, but they no longer get delivered to the consumer.

Is there any way, in the meantime, to manually force resume the delivery loop?

[ledoyen]

@mdiasribeiro Sorry for the delay, I hope you did not loose interrest on the project though :smiley:

I was able to mimic the behavior you described using a simple Alpakka build (See 9b71d359eb1fa37a8cfe6040580f7d9e9afd3ae3 for details).

Please let me now if 4b5755d14a9b90f85653eacc186a3470ef8439ef fixes your issue (fix available on the last SNAPSHOT build).

[mdiasribeiro]

@mdiasribeiro Sorry for the delay, I hope you did not loose interrest on the project though 😃

I was able to mimic the behavior you described using a simple Alpakka build (See 9b71d35 for details).

Please let me now if 4b5755d fixes your issue (fix available on the last SNAPSHOT build).

Thank you for the update! I will check if it resolves our issue and get back to you. Will be testing "1.0.12-SNAPSHOT" version, correct?

[ledoyen]

Yes, from the Sonatype repo.

[mdiasribeiro]

@ledoyen Thanks again for the update. It does fix part of our issue, so great, we can work around it! However, it does not quite work as we expected. We have the following set-up in tests:

1. Create a global MockConnectionFactory and a AmqpConnectionProvider out of it;
2. Pass it along to our system (with the alpakka streams set-up);
3. Use it to connect probes (producers and consumers) to do black-box testing;

Multiple connections are created using the AmqpConnectionProvider. In particular, seems like tests that involve shutting down the whole system (to test recovery) result in the mock itself being closed. I would expect that closing one connection would not stop the mock (except maybe when all connections were stopped). I see some code in MockConnection (specifically on the close and abort functions) that if called would do this behavior, although I do not know where that would be triggered from. It does force us to rebind the consumers after one connection gets closed. I might also be wrong on my diagnosis. Any idea?

[ledoyen]

Yes you are right, closing connection should not stop the mock. The root state of the mock (MockNode) is stored in the ConnectionFactory object.

Could you provide a sample project demonstrating the behavior you describe ? You can modify the setup https://github.com/fridujo/rabbitmq-mock/tree/master/src/it/alpakka if that helps.

Thanks in advance,

[mdiasribeiro]

Hi there, sorry about the delay. Managed to make a small reproducible example of one failure situation, although I suspect it might not be the only behavior I'm experiencing.

You can find it on the repository: https://github.com/mdiasribeiro/rabbitmq-mock-bugreport

I've managed to reduce it to not even using akka-streams at all. It seems that akka-streams ends up killing the connection on the producer side (connection.close(), possibly abort). I'm seeing that a separate consumer, using a different connection, gets unbound from the queue, which I find unexpected.

[ledoyen]

Hi, thanks for taking the time to supply a test to work with.

I will try to fix that asap (this week I hope).

[ledoyen]

@mdiasribeiro It should be working fine now, thanks again for providing a test about this issue.

Check-out version 1.0.12 and let me know !

[mdiasribeiro]

@ledoyen Thanks for the update! Preliminary testing looks promising and seems to show that our issues have been fixed. I'll be running tests over the week to confirm and ensure there are no further issues. Thanks!