[INLONG-8629][Agent] Fix sending invalid data to dataproxy failed blo…

…cks normal data sending (#8630)
apache · Aug 6, 2023 · be94dce · be94dce
1 parent 2fc03b4
commit be94dce
Showing 1 changed file with 19 additions and 0 deletions.
diff --git a/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java b/...agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/SenderManager.java
@@ -168,6 +168,21 @@ public void Stop() {
         shutdown = true;
         resendExecutorService.shutdown();
         sender.close();
+        cleanResendQueue();
+    }
+
+    private void cleanResendQueue() {
+        while (!resendQueue.isEmpty()) {
+            try {
+                AgentSenderCallback callback = resendQueue.poll(1, TimeUnit.SECONDS);
+                if (callback != null) {
+                    MemoryManager.getInstance()
+                            .release(AGENT_GLOBAL_WRITER_PERMIT, (int) callback.batchMessage.getTotalSize());
+                }
+            } catch (InterruptedException e) {
+                LOGGER.error("clean resend queue error{}", e.getMessage());
+            }
+        }
     }
 
     private AgentMetricItem getMetricItem(Map<String, String> otherDimensions) {
@@ -222,6 +237,10 @@ private void sendBatchWithRetryCount(BatchProxyMessage batchMessage, int retry)
         boolean suc = false;
         while (!suc) {
             try {
+                if (!resendQueue.isEmpty()) {
+                    AgentUtils.silenceSleepInMs(retrySleepTime);
+                    continue;
+                }
                 sender.asyncSendMessage(new AgentSenderCallback(batchMessage, retry),
                         batchMessage.getDataList(), batchMessage.getGroupId(), batchMessage.getStreamId(),
                         batchMessage.getDataTime(), SEQUENTIAL_ID.getNextUuid(), maxSenderTimeout, TimeUnit.SECONDS,