LogAggregationFileControllerFactory { public LogAggregationFileControllerFactory(Configuration conf) { this.conf = conf; Collection<String> fileControllers = conf.getStringCollection( YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS); List<String> controllerClassName = new ArrayList<>(); Map<String, String> controllerChecker = new HashMap<>(); for (String fileController : fileControllers) { Preconditions.checkArgument(validateAggregatedFileControllerName( fileController), "The FileControllerName: " + fileController + " set in " + YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS +" is invalid." + "The valid File Controller name should only " + "contain a-zA-Z0-9_ and can not start with numbers"); String remoteDirStr = String.format( YarnConfiguration.LOG_AGGREGATION_REMOTE_APP_LOG_DIR_FMT, fileController); String remoteDir = conf.get(remoteDirStr); boolean defaultRemoteDir = false; if (remoteDir == null || remoteDir.isEmpty()) { remoteDir = conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR); defaultRemoteDir = true; } String suffixStr = String.format( YarnConfiguration.LOG_AGGREGATION_REMOTE_APP_LOG_DIR_SUFFIX_FMT, fileController); String suffix = conf.get(suffixStr); boolean defaultSuffix = false; if (suffix == null || suffix.isEmpty()) { suffix = conf.get(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX, YarnConfiguration.DEFAULT_NM_REMOTE_APP_LOG_DIR_SUFFIX); defaultSuffix = true; } String dirSuffix = remoteDir + "-" + suffix; if (controllerChecker.containsKey(dirSuffix)) { if (defaultRemoteDir && defaultSuffix) { String fileControllerStr = controllerChecker.get(dirSuffix); List<String> controllersList = new ArrayList<>(); controllersList.add(fileControllerStr); controllersList.add(fileController); fileControllerStr = StringUtils.join(controllersList, ","); controllerChecker.put(dirSuffix, fileControllerStr); } else { String conflictController = controllerChecker.get(dirSuffix); throw new RuntimeException("The combined value of " + remoteDirStr + " and " + suffixStr + " should not be the same as the value" + " set for " + conflictController); } } else { controllerChecker.put(dirSuffix, fileController); } String classKey = String.format( YarnConfiguration.LOG_AGGREGATION_FILE_CONTROLLER_FMT, fileController); String className = conf.get(classKey); if (className == null || className.isEmpty()) { throw new RuntimeException("No class configured for " + fileController); } controllerClassName.add(className); Class<? extends LogAggregationFileController> sClass = conf.getClass( classKey, null, LogAggregationFileController.class); if (sClass == null) { throw new RuntimeException("No class defined for " + fileController); } LogAggregationFileController s = ReflectionUtils.newInstance( sClass, conf); if (s == null) { throw new RuntimeException("No object created for " + controllerClassName); } s.initialize(conf, fileController); controllers.add(s); } } LogAggregationFileControllerFactory(Configuration conf); LogAggregationFileController getFileControllerForWrite(); LogAggregationFileController getFileControllerForRead(
      ApplicationId appId, String appOwner); @Private @VisibleForTesting LinkedList<LogAggregationFileController> getConfiguredLogAggregationFileControllerList();  }
@Test public void testLogAggregationFileControllerFactory() throws Exception { enableFileControllers(ALL_FILE_CONTROLLERS, ALL_FILE_CONTROLLER_NAMES); LogAggregationFileControllerFactory factory = new LogAggregationFileControllerFactory(getConf()); List<LogAggregationFileController> list = factory.getConfiguredLogAggregationFileControllerList(); assertEquals("The expected number of LogAggregationFileController " + "is not 3!", 3, list.size()); assertTrue("Test format is expected to be the first " + "LogAggregationFileController!", list.get(0) instanceof TestLogAggregationFileController); assertTrue("IFile format is expected to be the second " + "LogAggregationFileController!", list.get(1) instanceof LogAggregationIndexedFileController); assertTrue("TFile format is expected to be the first " + "LogAggregationFileController!", list.get(2) instanceof LogAggregationTFileController); assertTrue("Test format is expected to be used for writing!", factory.getFileControllerForWrite() instanceof TestLogAggregationFileController); verifyFileControllerInstance(factory, TestLogAggregationFileController.class); }