Adds memory leak checking in unit test

brpc-java-communication/pom.xml

@@ -13,7 +13,6 @@
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   -->
-
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
@@ -33,6 +32,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <test.excludedGroups>com.baidu.brpc.test.IntegrationTest</test.excludedGroups>
     </properties>
 
     <dependencies>
@@ -44,10 +44,6 @@
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
         </dependency>
-<!--        <dependency>-->
-<!--            <groupId>com.google.protobuf</groupId>-->
-<!--            <artifactId>protobuf-java-util</artifactId>-->
-<!--        </dependency>-->
         <dependency>
             <groupId>com.googlecode.protobuf-java-format</groupId>
             <artifactId>protobuf-java-format</artifactId>
@@ -93,6 +89,7 @@
             <groupId>org.xerial.snappy</groupId>
             <artifactId>snappy-java</artifactId>
         </dependency>
+        <!-- region testing -->
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-slf4j-impl</artifactId>
@@ -113,7 +110,12 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
-
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <!-- endregion testing -->
         <!-- protostuff for stargate -->
         <dependency>
             <groupId>com.dyuproject.protostuff</groupId>
@@ -141,4 +143,26 @@
         </dependency>
     </dependencies>
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.12.4</version>
+                <configuration>
+                    <excludedGroups>${test.excludedGroups}</excludedGroups>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <profiles>
+        <profile>
+            <id>test-all</id>
+            <properties>
+                <test.excludedGroups/>
+            </properties>
+        </profile>
+    </profiles>
+
 </project>

brpc-java-communication/src/main/java/com/baidu/brpc/protocol/http/BrpcHttpObjectAggregator.java

@@ -15,6 +15,8 @@
  */
 package com.baidu.brpc.protocol.http;
 
+import com.baidu.brpc.exceptions.RpcException;
+
 import java.util.List;
 
 import io.netty.channel.ChannelHandlerContext;
@@ -46,4 +48,17 @@ public void aggregate(final ChannelHandlerContext ctx, Object msg, List<Object>
         }
     }
 
+    /**
+     * Abort aggregation and release underlying resources
+     * <p>
+     * FIXME find a better way to release memory
+     */
+    public void abort() {
+        try {
+            handlerRemoved(null);
+        } catch (Exception e) {
+            throw new RpcException(e);
+        }
+    }
+
 }

brpc-java-communication/src/main/java/com/baidu/brpc/protocol/http/BrpcHttpObjectDecoder.java

@@ -165,6 +165,7 @@ public Object decode(ChannelHandlerContext ctx, ByteBuf in) {
                 }
             }
             // decode failed, there's not enough bytes
+            httpObjectAggregator.abort();
             return null;
         } catch (DecoderException e) {
             throw e;

brpc-java-communication/src/test/java/com/baidu/brpc/protocol/http/BrpcHttpObjectDecoderTest.java

@@ -0,0 +1,68 @@
+package com.baidu.brpc.protocol.http;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.when;
+
+import com.baidu.brpc.test.DetectLeak;
+import com.baidu.brpc.test.DetectMemoryLeakRule;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.util.ReferenceCountUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.nio.charset.Charset;
+
+@Slf4j
+public class BrpcHttpObjectDecoderTest {
+
+    @DetectLeak
+    protected PooledByteBufAllocator alloc;
+    @Rule
+    public DetectMemoryLeakRule detectMemoryLeakRule = new DetectMemoryLeakRule(this);
+
+    @Test
+    public void testDecodePartial() throws Exception {
+        BrpcHttpObjectDecoder decoder = BrpcHttpObjectDecoder.getDecoder(true);
+        ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class);
+        when(ctx.alloc()).thenReturn(alloc);
+        ByteBuf buf = alloc.buffer(1024);
+        String[] testRequest = new String[]{
+                "GET / HTTP/1.1",
+                "Host: localhost",
+                "Content-Length: 4096",
+                "",
+                "abc"
+        }; // partial request
+        buf.writeBytes(StringUtils.join(testRequest, "\n\r").getBytes(Charset.forName("UTF-8")));
+        Object message = decoder.decode(ctx, buf);
+        assertThat(message).isNull();
+        ReferenceCountUtil.release(buf);
+    }
+
+    @Test
+    public void testDecode() throws Exception {
+        BrpcHttpObjectDecoder decoder = BrpcHttpObjectDecoder.getDecoder(true);
+        ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class);
+        when(ctx.alloc()).thenReturn(alloc);
+        ByteBuf buf = alloc.buffer(1024);
+        String[] testRequest = new String[]{
+                "GET / HTTP/1.1",
+                "Host: localhost",
+                "Content-Length: 10",
+                "",
+                "1234567890"
+        }; // partial request
+        buf.writeBytes(StringUtils.join(testRequest, "\n\r").getBytes(Charset.forName("UTF-8")));
+        Object message = decoder.decode(ctx, buf);
+        assertThat(message).isNotNull();
+        ReferenceCountUtil.release(buf);
+        ReferenceCountUtil.release(message);
+    }
+
+}

brpc-java-communication/src/test/java/com/baidu/brpc/test/DetectLeak.java

@@ -0,0 +1,11 @@
+package com.baidu.brpc.test;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.FIELD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface DetectLeak {
+}

brpc-java-communication/src/test/java/com/baidu/brpc/test/DetectMemoryLeakRule.java

@@ -0,0 +1,101 @@
+package com.baidu.brpc.test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.netty.buffer.PoolArenaMetric;
+import io.netty.buffer.PooledByteBufAllocator;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+@Slf4j
+public class DetectMemoryLeakRule implements TestRule {
+
+    private final Object instance;
+
+    public DetectMemoryLeakRule(Object instance) {
+        this.instance = instance;
+    }
+
+    @Override
+    public Statement apply(final Statement base, Description description) {
+        Class<?> testClass = description.getTestClass();
+        final List<Field> allocators = new ArrayList<Field>();
+        for (Field field : testClass.getDeclaredFields()) {
+            DetectLeak detectLeak = field.getAnnotation(DetectLeak.class);
+            if (detectLeak == null) {
+                continue;
+            }
+            if (!PooledByteBufAllocator.class.equals(field.getType())) {
+                continue;
+            }
+            field.setAccessible(true);
+            allocators.add(field);
+        }
+        return new Statement() {
+            @Override
+            public void evaluate() throws Throwable {
+                setupPools(allocators);
+                base.evaluate();
+                checkLeaks(allocators);
+            }
+        };
+    }
+
+    private void setupPools(List<Field> allocators) {
+        Iterator<Field> it = allocators.iterator();
+        while (it.hasNext()) {
+            Field field = it.next();
+            PooledByteBufAllocator alloc = new PooledByteBufAllocator(true, 0, 1, 8192, 11, 0, 0, 0, false);
+            try {
+                field.set(instance, alloc);
+            } catch (Exception ex) {
+                log.warn("Failed to process field {} for memory leak detection", field.getName());
+                it.remove();
+            }
+        }
+    }
+
+    private void checkLeaks(List<Field> allocators) {
+        for (Field field : allocators) {
+            PooledByteBufAllocator alloc;
+            try {
+                alloc = (PooledByteBufAllocator) field.get(instance);
+            } catch (Exception ex) {
+                log.warn("Failed to process field {} for memory leak detection", field.getName(), ex);
+                continue;
+            }
+            assertThat(alloc).as("PooledByteBufAllocator").isNotNull();
+            assertThat(getActiveHeapBuffers(alloc)).as("active heap memory").isZero();
+            assertThat(getActiveDirectBuffers(alloc)).as("active direct memory").isZero();
+        }
+    }
+
+    private static int getActiveDirectBuffers(PooledByteBufAllocator alloc) {
+        int directActive = 0, directAlloc = 0, directDealloc = 0;
+        for (PoolArenaMetric arena : alloc.metric().directArenas()) {
+            directActive += arena.numActiveAllocations();
+            directAlloc += arena.numAllocations();
+            directDealloc += arena.numDeallocations();
+        }
+        log.info("direct memory usage, active: {}, alloc: {}, dealloc: {}", directActive, directAlloc, directDealloc);
+        return directActive;
+    }
+
+    private static int getActiveHeapBuffers(PooledByteBufAllocator alloc) {
+        int heapActive = 0, heapAlloc = 0, heapDealloc = 0;
+        for (PoolArenaMetric arena : alloc.metric().heapArenas()) {
+            heapActive += arena.numActiveAllocations();
+            heapAlloc += arena.numAllocations();
+            heapDealloc += arena.numDeallocations();
+        }
+        log.info("heap memory usage, active: {}, alloc: {}, dealloc: {}", heapActive, heapAlloc, heapDealloc);
+        return heapActive;
+    }
+}

brpc-java-communication/src/test/java/com/baidu/brpc/test/IntegrationTest.java

@@ -0,0 +1,7 @@
+package com.baidu.brpc.test;
+
+/**
+ * Marker interface for Integration Test
+ */
+public interface IntegrationTest {
+}

brpc-java-communication/src/test/resources/log4j2.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+    <Appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="Console"/>
+        </Root>
+    </Loggers>
+</Configuration>

brpc-java-examples/brpc-java-core-examples/src/main/java/com/baidu/brpc/example/standard/EchoServiceImpl.java

@@ -51,7 +51,7 @@ public Echo.EchoResponse echo(Echo.EchoRequest request) {
         String message = request.getMessage();
         Echo.EchoResponse response = Echo.EchoResponse.newBuilder()
                 .setMessage(message).build();
-        LOG.debug("EchoService.echo, request={}, response={}",
+        LOG.info("EchoService.echo, request={}, response={}",
                 request.getMessage(), response.getMessage());
 
         return response;

pom.xml

@@ -13,7 +13,6 @@
   ~ See the License for the specific language governing permissions and
   ~ limitations under the License.
   -->
-
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
@@ -384,7 +383,7 @@
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>
-                <version>4.12</version>
+                <version>4.13</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
@@ -393,6 +392,12 @@
                 <version>2.23.4</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>org.assertj</groupId>
+                <artifactId>assertj-core</artifactId>
+                <version>1.7.1</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 </project>