HDDS-10333. RocksDB logger not closed (apache#6200)

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/db/DBStoreBuilder.java

@@ -52,6 +52,7 @@
 import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedColumnFamilyOptions;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedDBOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedLogger;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedRocksDB;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedStatistics;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedWriteOptions;
@@ -405,12 +406,7 @@ private ManagedDBOptions getDefaultDBOptions(
 
     // Apply logging settings.
     if (rocksDBConfiguration.isRocksdbLoggingEnabled()) {
-      org.rocksdb.Logger logger = new org.rocksdb.Logger(dbOptions) {
-        @Override
-        protected void log(InfoLogLevel infoLogLevel, String s) {
-          ROCKS_DB_LOGGER.info(s);
-        }
-      };
+      ManagedLogger logger = new ManagedLogger(dbOptions, (infoLogLevel, s) -> ROCKS_DB_LOGGER.info(s));
       InfoLogLevel level = InfoLogLevel.valueOf(rocksDBConfiguration
           .getRocksdbLogLevel() + "_LEVEL");
       logger.setInfoLogLevel(level);

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

@@ -18,20 +18,34 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.ratis.util.UncheckedAutoCloseable;
 import org.rocksdb.DBOptions;
+import org.rocksdb.Logger;
 
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.LOG;
 import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed DBOptions.
  */
 public class ManagedDBOptions extends DBOptions {
+
   private final UncheckedAutoCloseable leakTracker = track(this);
+  private final AtomicReference<Logger> loggerRef = new AtomicReference<>();
+
+  @Override
+  public DBOptions setLogger(Logger logger) {
+    IOUtils.close(LOG, loggerRef.getAndSet(logger));
+    return super.setLogger(logger);
+  }
 
   @Override
   public void close() {
     try {
+      IOUtils.close(LOG, loggerRef.getAndSet(null));
       super.close();
     } finally {
       leakTracker.close();

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

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.utils.db.managed;
+
+import org.apache.ratis.util.UncheckedAutoCloseable;
+import org.rocksdb.InfoLogLevel;
+import org.rocksdb.Logger;
+
+import java.util.function.BiConsumer;
+
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
+
+/** Managed {@link Logger}. */
+public class ManagedLogger extends Logger {
+
+  private final UncheckedAutoCloseable leakTracker = track(this);
+  private final BiConsumer<InfoLogLevel, String> delegate;
+
+  public ManagedLogger(ManagedDBOptions dbOptions, BiConsumer<InfoLogLevel, String> delegate) {
+    super(dbOptions);
+    this.delegate = delegate;
+  }
+
+  @Override
+  protected void log(InfoLogLevel infoLogLevel, String logMsg) {
+    delegate.accept(infoLogLevel, logMsg);
+  }
+
+  @Override
+  public void close() {
+    try {
+      super.close();
+    } finally {
+      leakTracker.close();
+    }
+  }
+}