(inlong) branch master updated: [INLONG-11175][Agent] Fix the problem of mqttsource message loss (#11176)

[dockerzhang]

This is an automated email from the ASF dual-hosted git repository.

dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new 43c3259f99 [INLONG-11175][Agent] Fix the problem of mqttsource message 
loss (#11176)
43c3259f99 is described below

commit 43c3259f998da11bb766cac7f1de2570f9f908a8
Author: Haotian Ma <60374114+qy-liu...@users.noreply.github.com>
AuthorDate: Tue Sep 24 12:07:22 2024 +0800

    [INLONG-11175][Agent] Fix the problem of mqttsource message loss (#11176)
---
 .../main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git 
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java
 
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java
index 144a1e6cc4..8ace035b7a 100644
--- 
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java
+++ 
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/MqttSource.java
@@ -115,7 +115,7 @@ public class MqttSource extends AbstractSource {
                     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);
+                    mqttMessagesQueue.offer(new DefaultMessage(recordValue, 
headerMap));
 
                 }