Stop dispatching when Consumer#close() is called

[hanzo2001]

I have a test Consumer that reads from a stream with 10000 messages.
I understand that the ::close operation will take some time before it is applied. However, when I call the close() method, the consumer will run for another ~400ms.

After attempting to close, the consumer sporadically reports every 40-60 messages with the same debug line.

... keeps consuming messages ...
DEBUG [pool-3-thread-1] c.r.s.i.ConsumersCoordinator   - Could not find stream subscription 0 or subscription closing, not providing credits
... keeps consuming messages ...

Eventually, the client does close

... keeps consuming messages ...
DEBUG [Thread-0] c.rabbitmq.stream.impl.Client  - Closing client
DEBUG [Thread-0] c.r.s.i.ConsumersCoordinator   - Disposed client subscription pool on ws-014:5552 because it was empty
DEBUG [Thread-0] c.rabbitmq.stream.impl.Client  - Client closed
DEBUG [Thread-0] c.r.s.i.ConsumersCoordinator   - Disposed client subscription pool on ws-014:5552 because it was empty
DEBUG [Thread-0] c.r.stream.impl.StreamConsumer - Calling tracking consumer closing callback (may be no-op)
DEBUG [Thread-0] c.r.stream.impl.StreamConsumer - Closed consumer successfully
DEBUG [Thread-0] c.r.stream.impl.StreamConsumer - Closed consumer successfully

Is there a way to control this delay?
The log shows duplicate lines. I don't know if this is normal.

[acogoluegnes]

These are in-flight chunks of messages still arriving to the closed consumer and the coordinator still provides credits after it received. That's a bug, even though it's harmless.

I doubt it makes the consumer closing longer though, but we'll see.

Thanks for reporting this.

[hanzo2001]

I am worried that my message handler is still being called even after I have called ::close(). Shouldn't the handler be ignored?
Can these in-flight chunks of messages be ignored?

I don't know if this should be part of a Discussion but ... How do we know that the consumer is closed? Is there a programmatic way of hooking a process after the consumer is done?

I understand that you are experimenting with the SubscriptionListener. Would this also be a place to run an onUnsubscribe event?

[acogoluegnes]

These are in-flight chunks of messages still arriving to the closed consumer and the coordinator still provides credits after it received. That's a bug, even though it's harmless.

This is wrong actually, the consumer does not ask for more credits if it's being closed.

I am worried that my message handler is still being called even after I have called ::close(). Shouldn't the handler be ignored?
Can these in-flight chunks of messages be ignored?

I pushed a commit to stop dispatching messages as soon as Consumer#closed() is called.

You should not worry about these messages:

DEBUG [pool-3-thread-1] c.r.s.i.ConsumersCoordinator   - Could not find stream subscription 0 or subscription closing, not providing credits

They're meant for debugging otherwise they would be at the WARN level.

I don't know if this should be part of a Discussion but ... How do we know that the consumer is closed? Is there a programmatic way of hooking a process after the consumer is done?
I understand that you are experimenting with the SubscriptionListener. Would this also be a place to run an onUnsubscribe event?

You can make sure a consumer is closed when the Consumer#close() method returns. It's idempotent, so you can call it as many times as you want.

I don't know if the SubscriptionListener is the good place for the closing event. There may be a need for some kind of event or notification API, but we haven't had much feedback on use cases that would require it.

[hanzo2001]

I pushed a commit to stop dispatching messages as soon as Consumer#closed() is called.

I will test this immediately. I need to also confirm that the consumer has stopped as soon as it returns

You can make sure a consumer is closed when the Consumer#close() method returns.

Regarding the idea behind onUnsubscribe, if what you say is true about consumer::close I don't think it would be advantageous.

[hanzo2001]

This works wonderfully! Thank you!

• The handler is no longer called after the close
• onUnsubscribe does not seem necessary
• ::close() seems to work seemlessly

Should I close this thread or should you close this thread?

[acogoluegnes]

Cool, thanks for the feedback.

We'll keep this issue as it contains the history and the commit references it. I just changed its name for the changelog.

[acogoluegnes]

Fixed by 34204ec.