SuperStream Consumer comes up deaf and never recovers until restart

[rbramante-hv]

Hi folks,

We are using RabbitMQ 3.13.7 server with Java Stream SDK 0.17.0.

We are hitting an intermittent issue where our Consumer starts "deaf": the stream exists, there are events being written to the stream from a Producer, and Rabbit reports a stream-locator connection from the consumer but no associated stream-consumer connection.

The Consumer never seems to recover once it comes up in this state. It will sit there indefinitely without change until we bounce it.

As part of debugging we added an Environment.streamExists() check to hold off building the Consumer until the Environment reported the stream existing. We observed that even after the Environment reported the stream as existing, Rabbit client logging showed:

[SERVER] 2024-11-16 20:43:30.034 INFO [rabbitmq-stream-locator-connection-0] ServerFrameHandler Route returned error: 2 (STREAM_DOES_NOT_EXIST)

We also registered a subscriptionListener when building the Consumer. When the Consumer starts clean we receive an almost immediate callback to the subscriptionListener. Not surprisingly, when the Consumer starts deaf we never receive any callback.

Final note: this is a SuperStream with a single partition (we wanted to start with a SuperStream in case we ever needed to expand to multiple partitions in the future). I assume this is not a common configuration, or maybe it is not recommended.

Any tips for additional debugging or things to try are much appreciated.

[acogoluegnes]

Could you share the server logs and a standalone Java project that reproduces the issue? Thanks.

[apinette-hv]

@acogoluegnes We appreciate your response.

See the attached file for the server logs and the standalone Java project which reproduces the way we initialize the client. We have not observed this failure with this sample running standalone outside of k8s, but it does present the issue running with k8s.

Let us know if you need any more information. Once again, we appreciate the help.

DeafSuperStreamConsumerPackage.tar.gz

[rbramante-hv]

@acogoluegnes I don't suppose you had any thoughts on our sample app if there is something we should be doing that we are not, or something we shouldn't be doing that we are?

[michaelklishin]

@rbramante-hv RabbitMQ 3.13 is out of community support. We need a way to reproduce this against 4.0.x.

[michaelklishin]

This specifically means that unless you are willing to upgrade and demonstrate evidence of an issue there, any kind of help you might get from the core team is nothing more than our own interest in digging out a potential problem in this client.

Which is not a given and depends on how easy it is to reproduce this behavior. If it is not easy, I'm afraid we won't spend much time on it.

[michaelklishin]

I do not see any client logs in the archive.

My first hypothesis was that

Route returned error: 2 (STREAM_DOES_NOT_EXIST)

is a misleading message of a client that's failed to connect.

I then found

2024-11-20 00:01:58.526031+00:00 [debug] <0.2672.0> Creating subscription 0 to <<"x.cdc.raw">>, with offset specification first, properties #{<<"name">> => <<"raw_stream_processor">>}
2024-11-20 00:01:58.526178+00:00 [debug] <0.2672.0> osiris: initialising reader. Spec: first
2024-11-20 00:01:58.527558+00:00 [debug] <0.2672.0> Next offset for subscription 0 is 0
2024-11-20 00:01:58.527705+00:00 [debug] <0.2672.0> Distributing existing messages to subscription 0 on <<"x.cdc.raw">>
2024-11-20 00:01:58.529038+00:00 [debug] <0.2672.0> Subscription 0 on <<"x.cdc.raw">> is now at offset 0 with 0 message(s) distributed after subscription
2024-11-20 00:01:59.222598+00:00 [debug] <0.2011.0> Received heartbeat frame post auth

in the node log which correspond to a stream protocol connection. So this hypothesis is incorrect.

There are also some AMQP (1.0 or 0-9-1) connections:

2024-11-19 23:59:59.246359+00:00 [info] <0.2023.0> accepting AMQP connection <0.2023.0> (10.42.248.222:59308 -> 10.42.125.167:5672)
2024-11-19 23:59:59.257688+00:00 [info] <0.2023.0> connection <0.2023.0> (10.42.248.222:59308 -> 10.42.125.167:5672) has a client-provided name: replication-provenance-stream-processor-6f67fdc556-gwch8-amqp-0
2024-11-19 23:59:59.257803+00:00 [debug] <0.2023.0> Client address during authN phase: {0,0,0,0,0,65535,2602,32167}
2024-11-19 23:59:59.257886+00:00 [debug] <0.2023.0> User 'rabbitAdminUser' authenticated successfully by backend rabbit_auth_backend_internal
2024-11-19 23:59:59.262737+00:00 [info] <0.2023.0> connection <0.2023.0> (10.42.248.222:59308 -> 10.42.125.167:5672 - replication-provenance-stream-processor-6f67fdc556-gwch8-amqp-0): user 'rabbitAdminUser' authenticated and granted access to vhost '/'
2024-11-19 23:59:59.587534+00:00 [info] <0.2036.0> accepting AMQP connection <0.2036.0> (10.42.248.198:34302 -> 10.42.125.167:5672)
2024-11-19 23:59:59.600862+00:00 [info] <0.2036.0> connection <0.2036.0> (10.42.248.198:34302 -> 10.42.125.167:5672) has a client-provided name: version-expiration-consumer-service-659698b758-b9fbv-amqp-0
2024-11-19 23:59:59.600976+00:00 [debug] <0.2036.0> Client address during authN phase: {0,0,0,0,0,65535,2602,32167}
2024-11-19 23:59:59.601071+00:00 [debug] <0.2036.0> User 'rabbitAdminUser' authenticated successfully by backend rabbit_auth_backend_internal
2024-11-19 23:59:59.605095+00:00 [info] <0.2036.0> connection <0.2036.0> (10.42.248.198:34302 -> 10.42.125.167:5672 - version-expiration-consumer-service-659698b758-b9fbv-amqp-0): user 'rabbitAdminUser' authenticated and granted access to vhost '/'
2024-11-20 00:00:00.455552+00:00 [warning] <0.2005.0> Socket #Port<0.257> closed [<0.2005.0>]
2024-11-20 00:00:00.455710+00:00 [debug] <0.2005.0> rabbit_stream_reader terminating in state 'open' with reason 'normal'
2024-11-20 00:00:00.591517+00:00 [info] <0.2059.0> accepting AMQP connection <0.2059.0> (10.42.248.232:44408 -> 10.42.125.167:5672)
2024-11-20 00:00:00.604457+00:00 [info] <0.2059.0> connection <0.2059.0> (10.42.248.232:44408 -> 10.42.125.167:5672) has a client-provided name: replication-metadata-push-service-5b7fc98bcc-gvjmb-amqp-0
2024-11-20 00:00:00.604576+00:00 [debug] <0.2059.0> Client address during authN phase: {0,0,0,0,0,65535,2602,32167}
2024-11-20 00:00:00.604680+00:00 [debug] <0.2059.0> User 'rabbitAdminUser' authenticated successfully by backend rabbit_auth_backend_internal
2024-11-20 00:00:00.608930+00:00 [info] <0.2059.0> connection <0.2059.0> (10.42.248.232:44408 -> 10.42.125.167:5672 - replication-metadata-push-service-5b7fc98bcc-gvjmb-amqp-0): user 'rabbitAdminUser' authenticated and granted access to vhost '/'
2024-11-20 00:00:00.843588+00:00 [info] <0.2075.0> accepting AMQP connection <0.2075.0> (10.42.248.233:53684 -> 10.42.125.167:5672)
2024-11-20 00:00:00.854689+00:00 [info] <0.2075.0> connection <0.2075.0> (10.42.248.233:53684 -> 10.42.125.167:5672) has a client-provided name: replication-object-data-pull-service-8668fb4c98-82fs4-amqp-0
2024-11-20 00:00:00.854812+00:00 [debug] <0.2075.0> Client address during authN phase: {0,0,0,0,0,65535,2602,32167}
2024-11-20 00:00:00.854913+00:00 [debug] <0.2075.0> User 'rabbitAdminUser' authenticated successfully by backend rabbit_auth_backend_internal
2024-11-20 00:00:00.858834+00:00 [info] <0.2075.0> connection <0.2075.0> (10.42.248.233:53684 -> 10.42.125.167:5672 - replication-object-data-pull-service-8668fb4c98-82fs4-amqp-0): user 'rabbitAdminUser' authenticated and granted access to vhost '/'
2024-11-20 00:00:01.413192+00:00 [warning] <0.2014.0> Socket #Port<0.259> closed [<0.2014.0>]
2024-11-20 00:00:01.413345+00:00 [debug] <0.2014.0> rabbit_stream_reader terminating in state 'open' with reason 'normal'

[michaelklishin]

@acogoluegnes the only thing in the server log that has attracted my attention is a couple of warnings:

2024-11-20 00:01:18.653001+00:00 [warning] <0.2437.0> Binding error: operation not permitted on the default exchange

There is also this warning that is likely coming from exchange federation, in which case it is very very unlikely to be relevant:

2024-11-19 23:59:40.617388+00:00 [warning] <0.1453.0> Binding error: no exchange 'x.lifecycle.policy' in vhost '/'

There are no errors logged of any kind.

#630 is my best guess given a quick inspection of the logs of this unsupported (without a commercial license) version.

[michaelklishin]

@rbramante-hv does the consumer ever receive any deliveries or there are none, ever on Kubernetes and only on Kubernetes? Have you triple-checked the topology and publishing settings?

[rbramante-hv]

@michaelklishin k8s is our primary deployment and test environment, so unfortunately the vast majority of runtime is there. We really only ran outside of k8s a handful of times trying to put together this sample test application for you and unfortunately we did not observe the same failure in that limited scope.

Once entered into this state, the consumer stays "deaf" and does not receive any events unless we bounce its service.

We believe the publishing settings to be correct because if we re-start the consumer it will pick up the events we expected it to receive, so they are present in the stream.

[michaelklishin]

@rbramante-hv there are at least three issues with your application code mentioned in #656 (reply in thread).

We will not troubleshoot your K8S deployment. That's on you. Both the server and the client are open source.

[michaelklishin]

Having just updated the release notes of RabbitMQ 4.1.0-beta.3, I wonder if this could potentially be rabbitmq/rabbitmq-server#12785, or at least it could have a similar effect on stream consumers.

@kjnilsson does the description here look familiar to what you've fixed in Osiris 1.8.5?

[acogoluegnes]

I haven't forgotten about this issue, it is on my todo list, but I haven't had time to look at it.

[acogoluegnes]

Thanks for the sample project, I had a look at it:

• please use the latest version of the stream Java client (bump it from 0.17.0 to 0.20.0)
• the code that checks the super stream exists is incorrect, and so is the code that creates it (it creates a regular stream). You should try to create the super stream (with the correct Java API), the operation is cheap enough to do on application start and is idempotent.

I cannot draw any conclusion from the server log. You can update the sample project with a publisher and some events that happen on the K8S environment (e.g. restarting the Docker container), this could help reproduce the problem.

[rbramante-hv]

@acogoluegnes Thanks for the reply. Sorry for the confusion; the create/exists check in the sample app was meant to act as a helper for standalone run but obviously was a source of confusion. In our testing, we were not directly creating the stream resource, it is being created by the Rabbit k8s messaging-topology-operator so the app was just coming online and using it. The definitions we are using look like:

queues:
  - name: EXAMPLE
    queueName: acme.EXAMPLE
    queueType: stream
    type: example

exchanges:
  - name: EXAMPLE
    exchangeName: acme.EXAMPLE
    type: direct
    arguments:
      x-super-stream: true

bindings:
  - name: EXAMPLE.binding
    source: acme.EXAMPLE
    sourceType: direct
    destination: acme.EXAMPLE
    destinationType: queue
    routingKey: "EXAMPLE"
    type: example
    arguments:
      x-stream-partition-order: 0

policies:
  - name: EXAMPLE.policy
    policyName: EXAMPLE.policy
    pattern: '^acme.EXAMPLE\$'
    applyTo: "streams"
    priority: 0
    arguments:
      queue-leader-locator: balanced

Please let us if you see something suspect with those. When we get a chance we can try upgrading the client and also testing against a 4.1 Rabbit cluster. Also will try to see if we can capture some additional logging from a k8s run when possible. Thanks again for all the insight.

[acogoluegnes]

I don't see anything obviously wrong with your settings. The stream name is not what it would have been if it had been created by the system, but that should not be a problem. Nevertheless, I would recommend following the convention. Create the super stream on a test environment with the CLI, e.g. rabbitmq-streams add_super_stream test --partitions 1, check out the created topology and make sure to create the exact same one on your environment. With the example above the partition stream would be named test-0.

I would also get rid of the 1-partition super stream in favor of a regular stream. It is not super streams do not work, but they do add complexity that will not help to diagnose the issue.

You should also try to reproduce the issue on a test K8S environment. You would be able to test with supported RabbitMQ versions and even with alphas. There is not certainty, but you could be hitting the issue @michaelklishin mentioned and the fix will no make it to 3.13.x unless you have a commercial license.

[michaelklishin]

To clarify, we will produce 3.13.x releases but only for users with a commercial license, and they will be distributed via the Broadcom customer portal, repositories. There won't be public binary releases on GitHub.