How to Increase chunk size for stream messages ingressed via mqtt plugin

[danieljirausch]

Community Support Policy

• I have read RabbitMQ's Community Support Policy
• I run RabbitMQ 4.x, the only series currently covered by community support
• I promise to provide all relevant information (versions, logs from all nodes, rabbitmq-diagnostics output, detailed reproduction steps)

RabbitMQ version used

4.1.2

Erlang version used

27.3.x

Operating system (distribution) used

Linux

How is RabbitMQ deployed?

Bitnami Helm chart

rabbitmq-diagnostics status output

not applicable

Logs from node 1 (with sensitive values edited out)

not applicable

Logs from node 2 (if applicable, with sensitive values edited out)

No response

Logs from node 3 (if applicable, with sensitive values edited out)

No response

rabbitmq.conf

for config see kubernetes section

Steps to deploy RabbitMQ cluster

Helm deploy via pipeline

Steps to reproduce the behavior in question

Messages are sent to RabbitMQ queues via MQTT message providers. Utilizing their routing keys they are eventually forwarded to streams.

The workflow is basically as follows:

• MQTT message provider sends message
• message is published in topic exchange "amq.topic" (RabbitMQ with MQTT Plugin)
• "amq.topic" is bound to another exchange (stream exchange)
  • type: x-modulus-hash
  • arguments: super-stream: true
  • routing-key for binding - basically everything
• streams are super stream partitions and bindings are created based on mod logic

advanced.config

No response

Application code

No response

Kubernetes deployment file

Kubernetes values yaml

---
image:
  registry: ...
  repository: docker.io-bitnamilegacy-rabbitmq
  tag: 4.1.3-debian-12-r1

global:
  security:
    allowInsecureImages: true

auth:
  username: ...
  existingPasswordSecret: ...
  existingErlangSecret: ...

extraPlugins: rabbitmq_auth_backend_oauth2 rabbitmq_mqtt rabbitmq_stream rabbitmq_stream_management rabbitmq_sharding

featureFlags: detailed_queues_endpoint,khepri_db,message_containers_deaths_v2,quorum_queue_non_voters,rabbit_exchange_type_local_random,rabbitmq_4.0.0,rabbitmq_4.1.0

resources:
  requests:
    cpu: 2
    memory: 2Gi

topologySpreadConstraints:
  - maxSkew: 1
    topologyKey: ...
    whenUnsatisfiable: DoNotSchedule

replicaCount: 3

metrics:
  enabled: true

service:
  type: LoadBalancer
  loadBalancerIP: ...
  extraPorts:
    - name: mqtt
      port: 1883
      targetPort: 1883
    - name: stream
      port: 5552
      targetPort: 5552

ingress:
  enabled: true
  hostname: ...
  ingressClassName: nginx

networkPolicy:
  extraIngress:
    - ports:
        - port: 6000  # Stream replication
          endPort: 6500
          protocol: TCP
      from:
        - podSelector:
            matchLabels:
              app.kubernetes.io/instance: rabbitmq-stage
              app.kubernetes.io/name: rabbitmq
    - ports:
        - protocol: TCP
          port: 1883  # MQTT port
        - protocol: TCP
          port: 5552  # Stream port
      from:
        - ipBlock:
            cidr: 0.0.0.0/0

persistence:
  enabled: true
  storageClass: tanzu-stretched-latebinding
  size: 25Gi

extraConfiguration: |-
  disk_free_limit.absolute = 4GB

  # --- Authentication ---
  auth_backends.1 = rabbit_auth_backend_internal
  auth_backends.2 = rabbit_auth_backend_oauth2

  # --- Management ---
  management.load_definitions = /app/load_definition.json
  management.oauth_enabled          = true
  management.oauth_client_id        = rabbitmq-ui
  auth_oauth2.additional_scopes_key = tags

  # --- OAuth2 Plugin Config ---
  auth_oauth2.resource_server_id    = rabbitmq
  auth_oauth2.issuer                = ...

  # --- TLS Security Disabled ---
  auth_oauth2.https.verify                = verify_none
  auth_oauth2.https.hostname_verification = none

  # --- MQTT ---
  mqtt.listeners.tcp.default = 1883
  mqtt.vhost                 = /
  mqtt.exchange              = amq.topic
  mqtt.allow_anonymous       = false

  # Use ETS (in memory) store for retained messages
  mqtt.retained_message_store = rabbit_mqtt_retained_msg_store_ets

  # --- Logging ---
  log.default.level = debug

extraSecrets:
  load-definition:
    load_definition.json: |+
      {
        "vhosts": [{ "name": "/" }],
        "policies": [
          {
            "vhost": "/",
            "name": "mqtt-default",
            "pattern": "^mqtt-subscription.*",
            "apply-to": "queues",
            "priority": 0,
            "definition": {
              "max-length": 1000000,
              "max-length-bytes": 1073741824
            }
          },
          {
            "vhost": "/",
            "name": "stream-default",
            "pattern": "^stream.dwh.*",
            "apply-to": "streams",
            "priority": 0,
            "definition": {
              "queue-leader-locator": "balanced",
              "max-length-bytes": 1073741824,
              "stream-max-segment-size-bytes": 104857600
            }
          }
        ]
      }

loadDefinition:
  enabled: true
  existingSecret: load-definition

What problem are you trying to solve?

Is there a way to increase the chunk size of messages stored in a stream, when they are ingressed via MQTT plugin? We suspect that this is the for slow performance.

More detail:
With the setup described in "Steps to reproduce the behavior in question" the messages can be successfully consumed by the clients.
So the message flow: MQTT > amq.topic exchange > super-stream exchange > stream partition > client works as desired.
What we noticed was, that with that setup the message throughput is rather slow. We used RabbitMQ Stream PerfTest and our internal opentelemetry metric to compare different setups and found that messages directly sent to a stream are consumed much faster than messages ingressed via MQTT.

The reason seems to be that messages ingressed via MQTT are stored in individual chunks. When consuming from a stream that holds these messages, the PerfTest tool always reports a chunk size of 1.

• Our messages are all similar in size and around 250 Bytes large.
• Current message ingress rate is about 1200 msg/s but will scale significantly

Throughput of our system when consuming from a stream that was directly filled with messages via stream producer (instead of MQTT plugin) is about on the level reported here:
https://www.rabbitmq.com/blog/2025/09/26/stream-delivery-optimization

[the-mikedavis]

The chunk size itself can't be tuned: the amount of data that ends up in a chunk is however much data the stream writer has in its mailbox at a point in time. So the higher the publishing throughput, the larger the chunk size.

Lower chunk sizes end up causing lower consumer throughput because the stream is being read in smaller sections. You should try upgrading to 4.2.0 as https://www.rabbitmq.com/blog/2025/09/26/stream-delivery-optimization specifically improves this scenario by reading ahead in the stream. I was looking into this for a somewhat similar use-case in #14877. In 4.2.1 you should be able to tune this read-ahead size to trade memory for higher consumer throughput.

I will try to reproduce this scenario with MQTT. I expect that 4.2.0 will make a big improvement.

[the-mikedavis]

I can't easily reproduce this naively (just using MQTT). It sounds like your topology distributes the messages to the different stream partitions so each stream has relatively low throughput and that leads to a small chunk size.

FYI you can use https://github.com/rabbitmq/omq to create synthetic workloads like you can with perf-test or stream-perf-test - that makes these scenarios much easier to reproduce.

[danieljirausch]

What would you consider as low message input?
I actually ran my tests also with directly binding streams to the topic exchange. Ingress rate was about 1k msg/s. That setup also resulted in chunks with size 1.

Thanks for suggesting omq. I will try it out and test some different scenarios.

[the-mikedavis]

I can't give exact numbers since it depends on the I/O resources used by RabbitMQ in your setup. In the discussion I linked, 250 MB/sec throughput had very low latency while 150 MB/sec throughput consumed slower and lead to high latency, but this was on another protocol and probably different hardware.

If you can make exact reproduction steps with omq (plus maybe streams-perf-test) I should be able to reproduce and see if the read-ahead optimizations from 4.2 improve this, and if the read-ahead default can be tweaked.

[acogoluegnes]

Chunk size is likely to stay 1 if ingress rate is in the 1,000s msg / s. You can try with 4.2 to see if the small-chunk optimization makes things better. You also mention that ingress rate "will scale significantly", so depending on the value you can end up with larger chunks and a better consuming rate.