HDDS-11543. Track OzoneClient object leaks via LeakDetector framework.

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java

@@ -877,4 +877,17 @@ public static HddsProtos.UUID toProtobuf(UUID uuid) {
         ? Thread.currentThread().getStackTrace()
         : null;
   }
+
+  /**
+   * Logs a warning to report that the class is not closed properly.
+   */
+  public static void reportLeak(Class<?> clazz, String stackTrace, Logger log) {
+    String warning = String.format("%s is not closed properly", clazz.getSimpleName());
+    if (stackTrace != null && LOG.isDebugEnabled()) {
+      String debugMessage = String.format("%nStackTrace for unclosed instance: %s",
+          stackTrace);
+      warning = warning.concat(debugMessage);
+    }
+    log.warn(warning);
+  }
 }

hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectUtils.java

@@ -55,12 +55,7 @@ static UncheckedAutoCloseable track(AutoCloseable object) {
 
   static void reportLeak(Class<?> clazz, String stackTrace) {
     ManagedRocksObjectMetrics.INSTANCE.increaseLeakObject();
-    String warning = String.format("%s is not closed properly", clazz.getSimpleName());
-    if (stackTrace != null && LOG.isDebugEnabled()) {
-      String debugMessage = String.format("%nStackTrace for unclosed instance: %s", stackTrace);
-      warning = warning.concat(debugMessage);
-    }
-    LOG.warn(warning);
+    HddsUtils.reportLeak(clazz, stackTrace, LOG);
   }
 
   private static @Nullable StackTraceElement[] getStackTrace() {

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneClient.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.ratis.util.UncheckedAutoCloseable;
 
 /**
  * OzoneClient connects to Ozone Cluster and
@@ -76,6 +77,7 @@ public class OzoneClient implements Closeable {
   private final ClientProtocol proxy;
   private final ObjectStore objectStore;
   private  ConfigurationSource conf;
+  private final UncheckedAutoCloseable leakTracker = OzoneClientFactory.track(this);
 
   /**
    * Creates a new OzoneClient object, generally constructed
@@ -119,7 +121,11 @@ public ConfigurationSource getConfiguration() {
    */
   @Override
   public void close() throws IOException {
-    proxy.close();
+    try {
+      proxy.close();
+    } finally {
+      leakTracker.close();
+    }
   }
 
   /**

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneClientFactory.java

@@ -23,9 +23,11 @@
 import java.io.IOException;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.conf.MutableConfigurationSource;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.LeakDetector;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.OzoneConsts;
@@ -34,13 +36,17 @@
 import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
 import org.apache.hadoop.security.token.Token;
 
-import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
-import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
-import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SERVICE_IDS_KEY;
+import org.apache.ratis.util.UncheckedAutoCloseable;
+
+import com.google.common.base.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SERVICE_IDS_KEY;
+
 /**
  * Factory class to create OzoneClients.
  */
@@ -54,6 +60,21 @@ public final class OzoneClientFactory {
    */
   private OzoneClientFactory() { }
 
+  private static final LeakDetector OZONE_CLIENT_LEAK_DETECTOR =
+      new LeakDetector("OzoneClientObject");
+
+  public static UncheckedAutoCloseable track(AutoCloseable object) {
+    final Class<?> clazz = object.getClass();
+    final StackTraceElement[] stackTrace = HddsUtils.getStackTrace(LOG);
+    return OZONE_CLIENT_LEAK_DETECTOR.track(object,
+        () -> HddsUtils.reportLeak(clazz,
+            HddsUtils.formatStackTrace(stackTrace, 4), LOG));
+  }
+
+  public static Logger getLogger() {
+    return LOG;
+  }
+
 
   /**
    * Constructs and return an OzoneClient with default configuration.

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/OzoneRpcClientTests.java

@@ -3140,6 +3140,32 @@ void testMultipartUploadOverride(ReplicationConfig replication)
     doMultipartUpload(bucket, keyName, (byte)97, replication);
 
   }
+
+  @Test
+  public void testClientLeakDetector() throws Exception {
+    OzoneClient client = OzoneClientFactory.getRpcClient(cluster.getConf());
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    String keyName = UUID.randomUUID().toString();
+    GenericTestUtils.LogCapturer ozoneClienFactoryLogCapturer =
+        GenericTestUtils.LogCapturer.captureLogs(
+            OzoneClientFactory.getLogger());
+
+    client.getObjectStore().createVolume(volumeName);
+    OzoneVolume volume = client.getObjectStore().getVolume(volumeName);
+    volume.createBucket(bucketName);
+    OzoneBucket bucket = volume.getBucket(bucketName);
+    byte[] data = new byte[10];
+    Arrays.fill(data, (byte) 1);
+    try (OzoneOutputStream out = bucket.createKey(keyName, 10,
+        ReplicationConfig.fromTypeAndFactor(RATIS, ONE), new HashMap<>())) {
+      out.write(data);
+    }
+    client = null;
+    System.gc();
+    GenericTestUtils.waitFor(() -> ozoneClienFactoryLogCapturer.getOutput()
+        .contains("is not closed properly"), 100, 2000);
+  }
   @Test
   public void testMultipartUploadOwner() throws Exception {
     // Save the old user, and switch to the old user after test