private List<AuditLogger> initAuditLoggers(Configuration conf) {
    // Initialize the custom access loggers if configured.
    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 {
          AuditLogger logger;
          if (DFS_NAMENODE_DEFAULT_AUDIT_LOGGER_NAME.equals(className)) {
            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) {
          throw new RuntimeException(e);
        }
      }
    }

    // Make sure there is at least one logger installed.
    if (auditLoggers.isEmpty()) {
      FSNamesystemAuditLogger fsNamesystemAuditLogger = new FSNamesystemAuditLogger();
      fsNamesystemAuditLogger.initialize(conf);
      auditLoggers.add(fsNamesystemAuditLogger);
    }

    // Add audit logger to calculate top users
    if (topConf.isEnabled && !topAuditLoggerAdded) {
      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);
      }
      auditLoggers.add(new TopAuditLogger(topMetrics));
    }

    return Collections.unmodifiableList(auditLoggers);
  }
@Test(timeout = 30000)
  public void testInitAuditLoggers() throws IOException {
    Configuration conf = new Configuration();
    FSImage fsImage = Mockito.mock(FSImage.class);
    FSEditLog fsEditLog = Mockito.mock(FSEditLog.class);
    Mockito.when(fsImage.getEditLog()).thenReturn(fsEditLog);
    FSNamesystem fsn;
    List<AuditLogger> auditLoggers;

    // Not to specify any audit loggers in config
    conf.set(DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY, "");
    // Disable top logger
    conf.setBoolean(DFSConfigKeys.NNTOP_ENABLED_KEY, false);
    conf.setBoolean(HADOOP_CALLER_CONTEXT_ENABLED_KEY, true);
    fsn = new FSNamesystem(conf, fsImage);
    auditLoggers = fsn.getAuditLoggers();
    assertTrue(auditLoggers.size() == 1);
    assertTrue(
        auditLoggers.get(0) instanceof FSNamesystem.FSNamesystemAuditLogger);
    FSNamesystem.FSNamesystemAuditLogger defaultAuditLogger =
        (FSNamesystem.FSNamesystemAuditLogger) auditLoggers.get(0);
    assertTrue(defaultAuditLogger.getCallerContextEnabled());

    // Not to specify any audit loggers in config
    conf.set(DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY, "");
    // Enable top logger
    conf.setBoolean(DFSConfigKeys.NNTOP_ENABLED_KEY, true);
    fsn = new FSNamesystem(conf, fsImage);
    auditLoggers = fsn.getAuditLoggers();
    assertTrue(auditLoggers.size() == 2);
    // the audit loggers order is not defined
    for (AuditLogger auditLogger : auditLoggers) {
      assertThat(auditLogger,
          either(instanceOf(FSNamesystem.FSNamesystemAuditLogger.class))
              .or(instanceOf(TopAuditLogger.class)));
    }

    // Configure default audit loggers in config
    conf.set(DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY, "default");
    // Enable top logger
    conf.setBoolean(DFSConfigKeys.NNTOP_ENABLED_KEY, true);
    fsn = new FSNamesystem(conf, fsImage);
    auditLoggers = fsn.getAuditLoggers();
    assertTrue(auditLoggers.size() == 2);
    for (AuditLogger auditLogger : auditLoggers) {
      assertThat(auditLogger,
          either(instanceOf(FSNamesystem.FSNamesystemAuditLogger.class))
              .or(instanceOf(TopAuditLogger.class)));
    }

    // Configure default and customized audit loggers in config with whitespaces
    conf.set(DFSConfigKeys.DFS_NAMENODE_AUDIT_LOGGERS_KEY,
        " default, org.apache.hadoop.hdfs.server.namenode.TestFSNamesystem$DummyAuditLogger  ");
    // Enable top logger
    conf.setBoolean(DFSConfigKeys.NNTOP_ENABLED_KEY, true);
    fsn = new FSNamesystem(conf, fsImage);
    auditLoggers = fsn.getAuditLoggers();
    assertTrue(auditLoggers.size() == 3);
    for (AuditLogger auditLogger : auditLoggers) {
      assertThat(auditLogger,
          either(instanceOf(FSNamesystem.FSNamesystemAuditLogger.class))
              .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));
  }