RabbitMQ Streams Go client latency test issue

[Kishan Sairam Adapa]

Hi,

I am trying to test latency of streams using the benchmark by tyler treat which was used to publish blog post Benchmarking Message Queue Latency.

In attempt to test streams, I've forked repository here where I've updated original repo to use go modules and implemented RabbitMQ streams (this file) to test. Upon running benchmark, I got a latency of 200 milliseconds, which is too high. And in management UI of RabbitMQ, stream producer and consumer network usages were topping at 5KB/s. 

I'm not sure if it's an issue with latency of go client or my implementation specifically. 

Instructions to reproduce:

- Ensure rabbitmq streams is running on 5552 localhost 

- Clone fork and download all deps by running go mod tidy

- cd into examples directory and `go run rmqstream.go`

Following is output of a run with each msg of size 1KB

``` 

~/repos/bench/examples$ go run rmqstream.go 

{Connections: 1, RequestRate: 1000, RequestTotal: 200, SuccessTotal: 200, ErrorTotal: 0, TimeElapsed: 39.999561743s, Throughput: 5.00/s}

```

Below is the graph obtained. Context on how benchmark operates with stream -> it sends a message to stream and receives it immediately (on same thread) and observes time difference.

[Gabriele Santomaggio]

Thank you for reporting.

I will have a look.

200 ms is the timeout for aggregate the messages, but I was planning to reduce it to 100 ms.

in the meantime there are two tests you can do:

use the batch-send interface that is synchronous and there is not aggregation logic, 

something like:

```

var arr []message.StreamMessage
for z := 0; z < bacthMessages; z++ {
   arr = append(arr, amqp.NewMessage([]byte("hello_world_"+strconv.Itoa(z))))
}
err := producer.BatchSend(arr) 

```

second test is to reduce the batch size:

```

env.NewProducer(streamName, stream.NewProducerOptions().SetBatchSize(XXXXX))

```

Cheers
-

Gabriele 

[Kishan Sairam Adapa]

Thank you for your response. I updated the benchmark with method you mentioned using batch size 1 (since benchmark doesn't consider batching).

Latency is reasonable now, attached graph comparing with traditional non durable rabbit mq queues. Increase in latency over queues given streams persistence seems reasonable to me.

So is aggregation a feature implemented by design to improve throughput / performance while trading off latency?

And I was wondering why client doesn't have consumer api return a channel that gets messages accumulated. For example Kafka Go clients do this, consumers return a channel which can be read from.

I do understand that user can declare a channel  themselves and send messages to channel inside the consumer handle message function.

But in my personal opinion the approach taken by other client libraries to return a channels seems more go-ish and cleaner (I might be wrong)

And a follow up, if we return a channel instead with consumer then will the aggregation timeout still come into play or the client doesn't wait to aggregate and sends messages as it receives?

[Gabriele Santomaggio]

I updated the example, see: https://github.com/Gsantomaggio/bench/tree/batch_send

So I moved from `RequestRate: 1000, RequestTotal: 200, SuccessTotal: 200,` to:

```

 go run rmqstream.go

{Connections: 1, RequestRate: 1000, RequestTotal: 15000, SuccessTotal: 15000, ErrorTotal: 0, TimeElapsed: 30.180111533s, Throughput: 497.02/s}

```

[Kishan Sairam Adapa]

Yes, indeed issue was with my implementation. There is 100x improvement in results. Could you also address my queries regarding why aggregation, consumer handler and why not channel based consumer API approach taken by other clients?

[Kishan Sairam Adapa]

My bad, there's no aggregation on consumer, producer aggregates messages to reduce network calls. Kindly ignore my query.

[Gabriele Santomaggio]

yes, the consumer does not aggregate the messages.

rabbitmq is push and kafka is pull.

Internally it uses a channel with the array of messages coming from the server: https://github.com/rabbitmq/rabbitmq-stream-go-client/blob/main/pkg/stream/client.go#L700

Thank you for your feedback btw

-

Gabriele