RATIS-2191. Add ResourceLeakDetector to Netty tests (#1180)

Author: ivandika3

ratis-examples/src/test/java/org/apache/ratis/examples/filestore/FileStoreStreamingBaseTest.java

@@ -56,6 +56,7 @@ public abstract class FileStoreStreamingBaseTest <CLUSTER extends MiniRaftCluste
         FileStoreStateMachine.class, StateMachine.class);
     ConfUtils.setFile(p::setFile, FileStoreCommon.STATEMACHINE_DIR_KEY,
         new File(getClassTestDir(), "filestore"));
+    DataStreamTestUtils.enableResourceLeakDetector();
   }
 
   static final int NUM_PEERS = 3;

ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamClusterTests.java

@@ -52,6 +52,7 @@ public abstract class DataStreamClusterTests<CLUSTER extends MiniRaftCluster> ex
     implements MiniRaftCluster.Factory.Get<CLUSTER> {
   {
     setStateMachine(MultiDataStreamStateMachine.class);
+    DataStreamTestUtils.enableResourceLeakDetector();
   }
 
   public static final int NUM_SERVERS = 3;

ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamTestUtils.java

@@ -46,6 +46,9 @@
 import org.apache.ratis.statemachine.TransactionContext;
 import org.apache.ratis.statemachine.impl.BaseStateMachine;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.io.netty.buffer.ByteBufUtil;
+import org.apache.ratis.thirdparty.io.netty.util.ResourceLeakDetector;
+import org.apache.ratis.thirdparty.io.netty.util.ResourceLeakDetector.Level;
 import org.apache.ratis.util.CollectionUtils;
 import org.apache.ratis.util.FileUtils;
 import org.apache.ratis.util.JavaUtils;
@@ -418,4 +421,13 @@ static void assertLogEntry(RaftServer.Division division, SingleDataStream stream
     final LogEntryProto entryFromLog = searchLogEntry(ClientInvocationId.valueOf(request), division.getRaftLog());
     Assertions.assertEquals(entryFromStream, entryFromLog);
   }
+
+  ResourceLeakDetector.LeakListener LEAK_LISTENER = (resourceType, records) -> {
+    throw new IllegalStateException("Leak detected for resource type: " + resourceType + records);
+  };
+
+  static void enableResourceLeakDetector() {
+    ResourceLeakDetector.setLevel(Level.PARANOID);
+    ByteBufUtil.setLeakListener(DataStreamTestUtils.LEAK_LISTENER);
+  }
 }