[Bug][Agent] The current MqttSource solution may cause message loss

[qy-liuhuo]

What happened

There is a problem in #10727

The current MqttSource solution may cause message loss

What you expected to happen

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    Map<String, String> headerMap = new HashMap<>();
    headerMap.put("record.topic", topic);
    headerMap.put("record.messageId", String.valueOf(message.getId()));
    headerMap.put("record.qos", String.valueOf(message.getQos()));
    byte[] recordValue = message.getPayload();
    mqttMessagesQueue.offer(new DefaultMessage(recordValue, headerMap), 1, TimeUnit.SECONDS);

}

Due to the timeout mechanism of blocking queue offers（mqttMessagesQueue.offer(new DefaultMessage(recordValue, headerMap), 1, TimeUnit.SECONDS);）, messages may be lost when the queue is full

How to reproduce

The mqtt production speed is higher than the consumption speed. When the blocking queue is filled, messages will be lost.

Environment

No response

InLong version

master

InLong Component

InLong Agent

Are you willing to submit PR?

• Yes, I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct