go-stream-locator

[Rozztig]

Describe the bug

Our service using rabbitmq-stream-go-client gets stuck at random times resulting in downtime. The issue starts when the RabbitMQ server logs "unknown command".

¤ unknown command {request,11,{close,1,<<79,75>>}}, closing connection. ¤ unknown command {response,18,{close,1}}, closing connection.

After seeing the unknown command our service fails to create new consumers. Instead we can see an increasing number of go-steam-locator in the RabbitMQ UI. This will go on until our service runs out of memory and restarts. So the issue is that the go-stream-locators get stuck forever. (Usually around 30-40k connections with only go-stream-locators before our service OOM)

Reproduction steps

1. Throttle RabbitMQ. (resource wise in Kubernetes)
2. Having a churn of 200/s opening and closing. NOTE: In production we have a churn around 3/s. Still happens once/twice a week
3. Look for unknown command in RabbitMQ logs.
4. Look in the RabbitMQ UI for stuck go-stream-locators ...

Expected behavior

For the go-stream-locators to vanish and messages going out to the consumers. Not hanging and taking about resources until crashing.

Additional context

We had this issue for 6 months now. We have decreased the frequency by lowering the opening and closing of connections. But it still happens randomly even with a lot of resources available. Using RabbitMQ 3.11.14 and rabbitmq-stream-go-client 1.1.2

[Gsantomaggio]

If you could provide a repository or source code to reproduce the issue would help to understand the problem. Thank you

[michaelklishin]

Or a Wireshark traffic capture and a set of exact steps used to "throttle RabbitMQ". Guessing is a very expensive way of troubleshooting distributed systems.

[Gsantomaggio]

Also, you should have some log client side. I am trying to reproduce the issue without success with:

    reader := bufio.NewReader(os.Stdin)
    env, err := stream.NewEnvironment(
        stream.NewEnvironmentOptions().
            SetVHost("/"))
    CheckErr(err)

    fmt.Print("Starting test")
    waitChan := make(chan struct{}, 1)
    for i := 0; i < 5000; i++ {
        waitChan <- struct{}{}
        if i%500 == 0 {
            fmt.Printf(" progress:%d \n", i)
        }
        go func() {
            streamName := uuid.New().String()
            err = env.DeclareStream(streamName, nil)
            CheckErr(err)
            producer, err := env.NewProducer(streamName, nil)
            CheckErr(err)
            producer.Close()
            consumer, err := env.NewConsumer(streamName, func(consumerContext stream.ConsumerContext, message *amqp.Message) {
                fmt.Printf("message received %s \n", message.GetData())
            }, nil)
            CheckErr(err)
            consumer.Close()
            _ = env.DeleteStream(streamName)
            <-waitChan
        }()
    }
    fmt.Print("Test completed")

[Rozztig]

If you could provide a repository or source code to reproduce the issue would help to understand the problem. Thank you

Thank you for your fast replies. Will try to create a example-project today that reproduces the issue.

During our testing we had throttled RabbitMQ in Kubernetes by setting the resource limits. Giving the pod a maximum of cpu: 500m memory: 2Gi This is just to trigger the issue faster. In production we give the pod 24Gi memory and cpu: 2 vcores. But we are nowhere near using that much resources and still experience the issue randomly.

But will reply later when we got a repo with an example reproducing the issue. Thanks!

[Gsantomaggio]

Closed by inactivity