From cd2c6b1aac470991b9b90339ce2721ba179e7c48 Mon Sep 17 00:00:00 2001
From: Ayush Saxena <ayushsaxena@apache.org>
Date: Thu, 27 Feb 2020 22:19:35 +0530
Subject: [PATCH] HDFS-15124. Crashing bugs in NameNode when using a valid
 configuration for . Contributed by Ctest.

---
 .../hadoop/hdfs/server/namenode/FSNamesystem.java | 10 +++++++++-
 .../hdfs/server/namenode/top/TopAuditLogger.java  | 15 +++++++++++++++
 .../hdfs/server/namenode/TestFSNamesystem.java    |  8 ++++++++
 3 files changed, 32 insertions(+), 1 deletion(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 8c5286cd36..ce44652774 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -1110,6 +1110,7 @@ private List<AuditLogger> initAuditLoggers(Configuration conf) {
     Collection<String> alClasses =
         conf.getTrimmedStringCollection(DFS_NAMENODE_AUDIT_LOGGERS_KEY);
     List<AuditLogger> auditLoggers = Lists.newArrayList();
+    boolean topAuditLoggerAdded = false;
     if (alClasses != null && !alClasses.isEmpty()) {
       for (String className : alClasses) {
         try {
@@ -1118,9 +1119,16 @@ private List<AuditLogger> initAuditLoggers(Configuration conf) {
             logger = new FSNamesystemAuditLogger();
           } else {
             logger = (AuditLogger) Class.forName(className).newInstance();
+            if (TopAuditLogger.class.getName().equals(
+                    logger.getClass().getName())) {
+              topAuditLoggerAdded = true;
+            }
           }
           logger.initialize(conf);
           auditLoggers.add(logger);
+        } catch (InstantiationException e) {
+          LOG.error("{} instantiation failed.", className, e);
+          throw new RuntimeException(e);
         } catch (RuntimeException re) {
           throw re;
         } catch (Exception e) {
@@ -1137,7 +1145,7 @@ private List<AuditLogger> initAuditLoggers(Configuration conf) {
     }
 
     // Add audit logger to calculate top users
-    if (topConf.isEnabled) {
+    if (topConf.isEnabled && !topAuditLoggerAdded) {
       topMetrics = new TopMetrics(conf, topConf.nntopReportingPeriodsMs);
       if (DefaultMetricsSystem.instance().getSource(
           TOPMETRICS_METRICS_SOURCE_NAME) == null) {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java
index 49c9153621..feba0e4f42 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/top/TopAuditLogger.java
@@ -23,12 +23,16 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.server.namenode.AuditLogger;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.hdfs.server.namenode.top.metrics.TopMetrics;
 
+import static org.apache.hadoop.hdfs.server.namenode.top.metrics.TopMetrics.TOPMETRICS_METRICS_SOURCE_NAME;
+
 /**
  * An {@link AuditLogger} that sends logged data directly to the metrics
  * systems. It is used when the top service is used directly by the name node
@@ -39,6 +43,17 @@ public class TopAuditLogger implements AuditLogger {
 
   private final TopMetrics topMetrics;
 
+  public TopAuditLogger() {
+    Configuration conf = new HdfsConfiguration();
+    TopConf topConf = new TopConf(conf);
+    this.topMetrics = new TopMetrics(conf, topConf.nntopReportingPeriodsMs);
+    if (DefaultMetricsSystem.instance().getSource(
+            TOPMETRICS_METRICS_SOURCE_NAME) == null) {
+      DefaultMetricsSystem.instance().register(TOPMETRICS_METRICS_SOURCE_NAME,
+              "Top N operations by user", topMetrics);
+    }
+  }
+
   public TopAuditLogger(TopMetrics topMetrics) {
     Preconditions.checkNotNull(topMetrics, "Cannot init with a null " +
         "TopMetrics");
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
index cb18bcf680..f2f4244024 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSNamesystem.java
@@ -303,6 +303,14 @@ public void testInitAuditLoggers() throws IOException {
               .or(instanceOf(TopAuditLogger.class))
               .or(instanceOf(DummyAuditLogger.class)));
     }
+
+    // Test Configuring TopAuditLogger.
+    conf.set(DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY,
+        "org.apache.hadoop.hdfs.server.namenode.top.TopAuditLogger");
+    fsn = new FSNamesystem(conf, fsImage);
+    auditLoggers = fsn.getAuditLoggers();
+    assertEquals(1, auditLoggers.size());
+    assertThat(auditLoggers.get(0), instanceOf(TopAuditLogger.class));
   }
 
   static class DummyAuditLogger implements AuditLogger {