MQTT input seemed not stable with high input load.

Dear Support,

I'm caught by a connection interruption issue during testing on mqtt input to the benthos.

To start with the test, I was using a small pieces of code as a MQTT client and publish messages to MQTT server (running in docker) every 5 second, then I can see the message though to the benthos output, where it connected to a Kafka instance. No error from Benthos side, and all messages seemed to landed to the output side.

Then I tried to increase the pushing interval of the MQTT client to one message per 20 ms, then I can see from the output side that seemed messages flow was interrupted, and then checking the log from Benthos container, I can see two lines of error messages also indicated that the connection was interrupted:

level=error msg="Connection lost for unknown reasons." @service=benthos label="" path=root.input.broker.inputs.1
level=error msg="Connection lost due to: pingresp not received, disconnecting" @service=benthos label="" path=root.input.broker.inputs.1

The connection seemed auto reestablished right after.
I know this error was popped from the underline lib used by Benthos, just wondering if there are some thing we could do from Benthos configuration side pls?

thanks!

Environment: Docker image with latest tag pull on 11th Oct, 2023.
MQTT server: EMQT
MQTT Client: with two senarios:

1. send a test message every 5 second, with 10 instances
2. send test message every 200 ms with one instance.

Both case will trigger the error.

Benthos: Config file:
input: broker: inputs: - mqtt: urls: - mqtt://xxxxx:1883 client_id: bentho_client1 user: xxxx password: xxxxx topics: - testtopic connect_timeout: 30s keepalive: 60

Start with the supplied MQTT input configuration and the two connection-loss log messages, then reproduce the interruption under the reported message rates. Compare the slow and high-load cases and determine whether Benthos configuration affects the disconnect; done means a validated configuration remedy or a documented limitation.