[incubator-inlong] branch master updated: [INLONG-3718][Audit] ByteBuf may leaks memory (#3725)

[he...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 82afdaa21 [INLONG-3718][Audit] ByteBuf may leaks memory (#3725)
82afdaa21 is described below

commit 82afdaa21bfd33660d8dd4721537ccedb1c9489e
Author: baomingyu <ba...@163.com>
AuthorDate: Fri Apr 15 14:35:41 2022 +0800

    [INLONG-3718][Audit] ByteBuf may leaks memory (#3725)
---
 .../src/main/java/org/apache/inlong/audit/send/SenderHandler.java    | 5 ++---
 .../src/main/java/org/apache/inlong/audit/send/SenderManager.java    | 5 ++---
 .../src/main/java/org/apache/inlong/audit/util/Decoder.java          | 4 ++--
 3 files changed, 6 insertions(+), 8 deletions(-)

diff --git a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
index 2a336f349..dd8ff1a3a 100644
--- a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
+++ b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderHandler.java
@@ -17,13 +17,12 @@
 
 package org.apache.inlong.audit.send;
 
-import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class SenderHandler extends SimpleChannelInboundHandler<ByteBuf> {
+public class SenderHandler extends SimpleChannelInboundHandler<byte[]> {
     private static final Logger logger = LoggerFactory.getLogger(SenderHandler.class);
     private SenderManager manager;
 
@@ -40,7 +39,7 @@ public class SenderHandler extends SimpleChannelInboundHandler<ByteBuf> {
      * Message Received
      */
     @Override
-    public void channelRead0(io.netty.channel.ChannelHandlerContext ctx, ByteBuf e)  {
+    public void channelRead0(io.netty.channel.ChannelHandlerContext ctx, byte[] e)  {
         try {
             manager.onMessageReceived(ctx, e);
         } catch (Throwable ex) {
diff --git a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
index 4355a6a09..d7c9139d9 100644
--- a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
+++ b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/send/SenderManager.java
@@ -268,11 +268,10 @@ public class SenderManager {
      * @param ctx ctx
      * @param msg msg
      */
-    public void onMessageReceived(ChannelHandlerContext ctx, ByteBuf msg) {
+    public void onMessageReceived(ChannelHandlerContext ctx, byte[] msg) {
         try {
             //Analyze abnormal events
-            ByteBuf readBuffer = msg;
-            byte[] readBytes = readBuffer.array();
+            byte[] readBytes = msg;
             AuditApi.BaseCommand baseCommand = AuditApi.BaseCommand.parseFrom(readBytes);
             // Parse request id
             Long requestId = baseCommand.getAuditReply().getRequestId();
diff --git a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
index 03ff94312..cf14e471f 100644
--- a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
+++ b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/util/Decoder.java
@@ -18,7 +18,6 @@
 package org.apache.inlong.audit.util;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToMessageDecoder;
 import java.util.List;
@@ -49,7 +48,8 @@ public class Decoder extends MessageToMessageDecoder<ByteBuf> {
             buffer.resetReaderIndex();
             return;
         }
-        ByteBuf returnBuffer = ByteBufAllocator.DEFAULT.buffer(totalLen);
+        byte[] returnBuffer = new byte[totalLen];
         buffer.readBytes(returnBuffer, 0, totalLen);
+        out.add(returnBuffer);
     }
 }