ClusterConfigManager { @Transactional public ClusterEntity createClusterConfig(ClusterCreate cluster) { String name = cluster.getName(); if (name == null || name.isEmpty()) { throw ClusterConfigException.CLUSTER_NAME_MISSING(); } List<String> failedMsgList = new ArrayList<String>(); List<String> warningMsgList = new ArrayList<String>(); String appManager = cluster.getAppManager(); if (appManager == null) { appManager = Constants.IRONFAN; } SoftwareManager softwareManager = getSoftwareManager(appManager); HadoopStack stack = filterDistroFromAppManager(softwareManager, cluster.getDistro()); if (cluster.getDistro() == null || stack == null) { throw BddException.INVALID_PARAMETER("distro", cluster.getDistro()); } ClusterBlueprint blueprint = cluster.toBlueprint(); try { softwareManager.validateBlueprint(cluster.toBlueprint()); cluster.validateClusterCreate(failedMsgList, warningMsgList); } catch (ValidationException e) { failedMsgList.addAll(e.getFailedMsgList()); warningMsgList.addAll(e.getWarningMsgList()); } if (!failedMsgList.isEmpty()) { throw ClusterConfigException.INVALID_SPEC(failedMsgList); } if (!validateRacksInfo(cluster, failedMsgList)) { throw ClusterConfigException.INVALID_PLACEMENT_POLICIES(failedMsgList); } String localRepoURL = cluster.getLocalRepoURL(); if (!CommonUtil.isBlank(localRepoURL) && !validateLocalRepoURL(localRepoURL)) { throw ClusterConfigException.INVALID_LOCAL_REPO_URL(failedMsgList); } try { ClusterEntity entity = clusterEntityMgr.findByName(name); if (entity != null) { logger.info("can not create cluster " + name + ", which is already existed."); throw BddException.ALREADY_EXISTS("Cluster", name); } updateInfrastructure(cluster, softwareManager, blueprint); logger.debug("begin to add cluster config for " + name); Gson gson = new Gson(); ClusterEntity clusterEntity = new ClusterEntity(name); clusterEntity.setAppManager(cluster.getAppManager()); clusterEntity.setDistro(cluster.getDistro()); clusterEntity.setDistroVendor(cluster.getDistroVendor()); clusterEntity.setDistroVersion(cluster.getDistroVersion()); clusterEntity.setStartAfterDeploy(true); clusterEntity.setPassword(cluster.getPassword()); clusterEntity.setTemplateId(this.nodeTemplateService.getNodeTemplateIdByName(cluster.getTemplateName())); clusterEntity.setVersion(clusterEntityMgr.getServerVersion()); if (softwareManager.containsComputeOnlyNodeGroups(blueprint)) { clusterEntity.setAutomationEnable(automationEnable); } else { clusterEntity.setAutomationEnable(null); } clusterEntity.setVhmMinNum(-1); clusterEntity.setVhmMaxNum(-1); if (cluster.getRpNames() != null && cluster.getRpNames().size() > 0) { logger.debug("resource pool " + cluster.getRpNames() + " specified for cluster " + name); clusterEntity.setVcRpNameList(cluster.getRpNames()); } else { logger.debug("no resource pool name specified, use global configuration."); } if (cluster.getDsNames() != null && !cluster.getDsNames().isEmpty()) { logger.debug("datastore " + cluster.getDsNames() + " specified for cluster " + name); clusterEntity.setVcDatastoreNameList(cluster.getDsNames()); } else { logger.debug("no datastore name specified, use global configuration."); } clusterEntity.setNetworkConfig(validateAndConvertNetNamesToNetConfigs( cluster.getNetworkConfig(), cluster.getDistroVendor() .equalsIgnoreCase(Constants.MAPR_VENDOR))); clusterEntity.setVhmJobTrackerPort("50030"); if (cluster.getConfiguration() != null && cluster.getConfiguration().size() > 0) { clusterEntity.setHadoopConfig((new Gson()).toJson(cluster .getConfiguration())); updateVhmJobTrackerPort(cluster, clusterEntity); } setInfraConfig(cluster, clusterEntity); setAdvancedProperties(cluster.getExternalHDFS(), cluster.getExternalMapReduce(), localRepoURL, cluster.getExternalNamenode(), cluster.getExternalSecondaryNamenode(), cluster.getExternalDatanodes(), cluster.getClusterCloneType(), clusterEntity); NodeGroupCreate[] groups = cluster.getNodeGroups(); if (groups != null && groups.length > 0) { clusterEntity.setNodeGroups(convertNodeGroupsToEntities(gson, clusterEntity, cluster.getDistro(), groups, cluster.isValidateConfig())); validateMemorySize(clusterEntity.getNodeGroups(), failedMsgList); if (!failedMsgList.isEmpty()) { throw ClusterConfigException.INVALID_SPEC(failedMsgList); } } if(!CommonUtil.isBlank(cluster.getAppManager()) && Constants.IRONFAN.equals(cluster.getAppManager())) for(NodeGroupEntity group : clusterEntity.getNodeGroups()){ String groupRoles = group.getRoles(); if((group.getLatencySensitivity() == LatencyPriority.HIGH) && ((groupRoles.contains(HadoopRole.HBASE_REGIONSERVER_ROLE.toString())))){ setHbaseRegionServerOpts(cluster, group); if (cluster.getConfiguration() != null && cluster.getConfiguration().size() > 0) { clusterEntity.setHadoopConfig((new Gson()).toJson(cluster .getConfiguration())); } break; } } if (cluster.getTopologyPolicy() == null) { clusterEntity.setTopologyPolicy(TopologyType.NONE); } else { clusterEntity.setTopologyPolicy(cluster.getTopologyPolicy()); } if (clusterEntity.getTopologyPolicy() == TopologyType.HVE) { boolean hveSupported = false; if (clusterEntity.getDistro() != null) { HadoopStack hadoopStack = filterDistroFromAppManager(softwareManager, clusterEntity.getDistro()); if (hadoopStack != null) { hveSupported = hadoopStack.isHveSupported(); } } if (!hveSupported) { throw ClusterConfigException.INVALID_TOPOLOGY_POLICY( clusterEntity.getTopologyPolicy(), "current Hadoop distribution does not support HVE."); } } clusterEntityMgr.insert(clusterEntity); logger.debug("finished to add cluster config for " + name); return clusterEntity; } catch (UniqueConstraintViolationException ex) { logger.info("can not create cluster " + name + ", which is already existed."); throw BddException.ALREADY_EXISTS(ex, "Cluster", name); } }  IDatastoreService getDatastoreMgr(); void setDatastoreMgr(IDatastoreService datastoreMgr); RackInfoManager getRackInfoMgr(); void setRackInfoMgr(RackInfoManager rackInfoMgr); IResourcePoolService getRpMgr(); void setRpMgr(IResourcePoolService rpMgr); INetworkService getNetworkMgr(); void setNetworkMgr(INetworkService networkMgr); IClusterEntityManager getClusterEntityMgr(); @Autowired void setClusterEntityMgr(IClusterEntityManager clusterEntityMgr); IClusteringService getClusteringService(); @Autowired void setClusteringService(IClusteringService clusteringService); @Autowired void setSoftwareManagerCollector(
         SoftwareManagerCollector softwareManagerCollector); @Transactional ClusterEntity createClusterConfig(ClusterCreate cluster); HadoopStack filterDistroFromAppManager(
         SoftwareManager softwareManager, String distroName); NodeGroupEntity convertGroup(Gson gson, ClusterEntity clusterEntity,
         NodeGroupCreate group, String distro, boolean validateWhiteList); @Transactional(readOnly = true) ClusterCreate getClusterConfig(String clusterName); @Transactional(readOnly = true) ClusterCreate getClusterConfig(String clusterName, boolean needAllocIp); @Transactional void updateAppConfig(String clusterName, ClusterCreate clusterCreate); @Transactional void validateRackTopologyUploaded(Set<String> hosts, String topology); @Transactional Map<String, String> buildTopology(List<NodeRead> nodes,
         String topology); SoftwareManager getSoftwareManager(String appManager);  }
@Test(groups = { "TestClusterConfigManager" }) public void testClusterConfigWithExternalMapReduce() throws Exception { String externalMR = "192.168.0.1:8021"; String externalHDFS = "hdfs: ClusterCreate spec = new ClusterCreate(); spec.setName("my-cluster-external-mr"); List<String> rps = new ArrayList<String>(); rps.add("myRp1"); spec.setRpNames(rps); spec.setNetworkConfig(createNetConfigs()); spec.setDistro("bigtop"); spec.setDistroVendor(Constants.DEFAULT_VENDOR); spec.setExternalMapReduce(externalMR); spec.setExternalHDFS(externalHDFS); String clusterConfigJson = "{\"configuration\":{\"hadoop\":{\"mapred-site.xml\":{\"mapred.job.tracker\":\"" + externalMR + "\"}, \"core-site.xml\":{\"fs.default.name\":\"" + externalHDFS + "\"}}}}"; Map clusterConfig = (new Gson()).fromJson(clusterConfigJson, Map.class); spec.setConfiguration((Map<String, Object>) (clusterConfig .get("configuration"))); NodeGroupCreate worker = new NodeGroupCreate(); List<String> computeRoles = new ArrayList<String>(); computeRoles.add("hadoop_nodemanager"); worker.setRoles(computeRoles); worker.setName("compute1"); worker.setInstanceNum(2); worker.setInstanceType(InstanceType.MEDIUM); StorageRead storage = new StorageRead(); storage.setType("LOCAL"); storage.setSizeGB(10); worker.setStorage(storage); spec.setNodeGroups(new NodeGroupCreate[] { worker }); spec = ClusterSpecFactory.getCustomizedSpec(spec, null); clusterConfigMgr.createClusterConfig(spec); ClusterEntity cluster = clusterEntityMgr.findByName("my-cluster-external-mr"); Assert.assertTrue(cluster != null); Assert.assertEquals( cluster.getAdvancedProperties(), "{\"ExternalMapReduce\":\"192.168.0.1:8021\",\"ExternalHDFS\":\"hdfs: ClusterRead clusterRead = clusterEntityMgr.toClusterRead("my-cluster-external-mr", true); Assert.assertEquals(clusterRead.getExternalHDFS(), "hdfs: Assert.assertEquals(clusterRead.getExternalMapReduce(), "192.168.0.1:8021"); }