Consumer state listener usage

[pvaiko]

Hello,

#836 added listeners to track the state of consumers or producers, which is a very welcome feature for our use case.

Our consumers must know whether they do not receive messages because producers don't send them, or because of a connection/broker problem.

We can implement some kind of lifesign on the producer/consumer level, but a state listener is obviously the preferred solution. My question is how exactly can I determine when the consumer is disconnected (that is, is not able to receive messages) and when is it healthy again.

The javadoc says that "A consumer uses one connection for consuming and another connection for server-side offset tracking", and indeed, they come in pairs for my test consumers:

[2026-02-01 07:25:24,182] DEBUG [main] [RabbitMqStreamConsumer-test-consumer-1] Resource ({ "consumer_id" : 0,"stream" : "test-stream","subscription_client" : "127.0.0.1:58582 -> localhost:5552", "tracking_client" : "127.0.0.1:58580 -> localhost:5552"}) state changed, was OPENING, now OPEN [{}]
[2026-02-01 07:25:32,634] DEBUG [main] [RabbitMqStreamConsumer-test-consumer-1] Resource ({ "consumer_id" : 1,"stream" : "test-stream","subscription_client" : "127.0.0.1:58582 -> localhost:5552", "tracking_client" : "127.0.0.1:58580 -> localhost:5552"}) state changed, was OPENING, now OPEN [{}]
...
[2026-02-01 07:25:57,802] DEBUG [main] [RabbitMqStreamConsumer-test-consumer-1] Resource ({ "consumer_id" : 0,"stream" : "test-stream","subscription_client" : "127.0.0.1:58582 -> localhost:5552", "tracking_client" : "127.0.0.1:58580 -> localhost:5552"}) state changed, was OPEN, now CLOSED [{}]
[2026-02-01 07:25:57,803] DEBUG [main] [RabbitMqStreamConsumer-test-consumer-1] Resource ({ "consumer_id" : 1,"stream" : "test-stream","subscription_client" : "127.0.0.1:58582 -> localhost:5552", "tracking_client" : "127.0.0.1:58580 -> localhost:5552"}) state changed, was OPEN, now CLOSED [{}]

My Resource.StateListener is called once with a Resource.Context for each of the two resources in case of my test Consumer (as the Javadoc says).

Now, conservatively, my application could decide that if at least one of the resources is not OPEN, then there is likely a connection issue. But there seem to be no way to programmatically find out how many resources are there, so after my listener is called with an OPEN state, I don't know how many more resources are there. If there is one, I can report to the application that the connection is up. If there is more, I'd have to wait until all resources are reported as OPEN, but again, I don't know how many more to expect.

I can track the resources in my consumer, for instance maintaining a map with the Resource as the key to detect all edge cases, but not knowing how many resources are there makes this fragile and possibly memory leaks as it would rely on the stream library to maintain Resources with the same hash during the lifetime of the consumer.

Is there something I'm missing? What is the best practice to detect consumer state changes?

Thanks

[acogoluegnes]

The consumer or the producer is the resource. So if there is a connection issue, you should see the state changing to "recovering" then going back to "open". A named consumer that uses server-side offset tracking uses an extra connection, so you'll see also events when this connection goes down and reconnects.

[acogoluegnes]

In the case of a consumer it is not possible to know which connection is down or up. The state listener API is decoupled from what it monitors, it is hard to have details without breaking this decoupling. We could provide a map with a "loose" contract for the entries, like "tracking" => "down", "subscription" => "up", etc.

What do you need these details for?

[pvaiko]

I don't need the details, what I need is one single state, which is exactly what you have now in the subscription state. If there are two states as now, and they are delivered in separate listener calls, and I don't know which state is which, then I have to implement some logic (based on assumptions) in the client to figure out, something like I wrote above:

that I keep a counter in my consumer and increment it by one whenever I receive an non-OPEN -> OPEN state transition and decrement by one on OPEN -> non-OPEN transitions, this way my consumer is healthy when this counter == 2.

To avoid this, I see to easy solutions:

• don't call the state listener with the tracking connection, only with the subscription connection, since that is now only OPEN when the tracking is OPEN as well, or
• include both states in each call of the Resource.StateListener, as a list of states (as opposed to a separate call for each Resource), this way the listener can determine if the consumer is healthy without keeping a state itself.

Thank you for your time.

[pvaiko]

To illustrate my point, this is my current listener:

    private void handleState(Resource.Context context) {
        logger.debug("Resource ({}) state changed, was {}, now {}",
                context.resource(), context.previousState(), context.currentState());
        
        var previousHealthyConnectionStateCount = this.resources.entrySet()
                .stream()
                .filter(e -> e.getValue() == Resource.State.OPEN)
                .count();

        this.resources.put(context.resource(), context.currentState());
        if (resources.size() > 2) {
            logger.warn("More than 2 resources in state listener, this is unexpected. Resources: {}", resources);
        }

        var currentHealthyConnectionStateCount = this.resources.entrySet()
                .stream()
                .filter(e -> e.getValue() == Resource.State.OPEN)
                .count();
        // with the stream lib <= 1.4.0, we must implement this logic as there are two states:
        // 1. tracking connection state, 2. subscription connection state, and we have no idea which one is the resource
        // we are being called for. Since we assume the connection is healthy only if we have both in the OPEN state,
        // we only notify our client when the number of OPEN connections changes from 2 to less than 2 or vice versa.
        if ((previousHealthyConnectionStateCount != 2) && (currentHealthyConnectionStateCount == 2)) {
            // both connections are healthy
            this.stateListener.onStateChange(StreamState.State.UP);
        } else if ((previousHealthyConnectionStateCount == 2) && (currentHealthyConnectionStateCount != 2)) {
            // at least one connection is down
            this.stateListener.onStateChange(StreamState.State.DOWN);
        }
    }
    

[acogoluegnes]

Again, the Consumer or Producer instance is the resource from Context#resource(). If you set up a StateListener instance on e.g. a Consumer, it should always see the same consumer instance coming from Context#resource(). If you don't observe this, there is a bug somewhere. In this case, provide a standalone code sample that reproduces this behavior.

Of course if you have a shared StateListener instance between consumers and producers (that is you create a StateListener instance and use it when creating your consumers and producers), it will see different resources in its callback, that is perfectly normal.

According to the comment in your code, the current behavior should do what you expect. For example with a named consumer, here are the events you should see once it is initialized:

• subscription connection goes down => event with "recovering"
• subscription connection goes back up => event with "open"
• subscription connection goes down => event with "recovering"
• tracking connection goes down => no event
• subscription connection goes back up => no event
• tracking connection goes back up => event with "open"

If this is what you expect and don't see, there is a bug, so provide a code sample and steps to reproduce the issue so I can fix it.

[pvaiko]

Thank you very much, all works fine, I don't see any bugs!