You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2024/09/24 04:07:27 UTC
(inlong) branch master updated: [INLONG-11175][Agent] Fix the problem of mqttsource message loss (#11176)
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 <60...@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));
}