MRRunningJobApplication extends StormApplication { @Override public StormTopology execute(Config config, StormEnvironment environment) { MRRunningJobConfig mrRunningJobConfig = MRRunningJobConfig.newInstance(config); String[] confKeyPatternsSplit = mrRunningJobConfig.getConfig().getString("MRConfigureKeys.jobConfigKey").split(","); List<String> confKeyKeys = new ArrayList<>(confKeyPatternsSplit.length); for (String confKeyPattern : confKeyPatternsSplit) { confKeyKeys.add(confKeyPattern.trim()); } confKeyKeys.add(Constants.JobConfiguration.CASCADING_JOB); confKeyKeys.add(Constants.JobConfiguration.HIVE_JOB); confKeyKeys.add(Constants.JobConfiguration.PIG_JOB); confKeyKeys.add(Constants.JobConfiguration.SCOOBI_JOB); confKeyKeys.add(0, mrRunningJobConfig.getConfig().getString("MRConfigureKeys.jobNameKey")); TopologyBuilder topologyBuilder = new TopologyBuilder(); String spoutName = "mrRunningJobFetchSpout"; String boltName = "mrRunningJobParseBolt"; int tasks = mrRunningJobConfig.getConfig().getInt("stormConfig." + spoutName + "Tasks"); topologyBuilder.setSpout( spoutName, new MRRunningJobFetchSpout(mrRunningJobConfig.getEndpointConfig(), mrRunningJobConfig.getZkStateConfig()), tasks ).setNumTasks(tasks); tasks = mrRunningJobConfig.getConfig().getInt("stormConfig." + boltName + "Tasks"); topologyBuilder.setBolt(boltName, new MRRunningJobParseBolt( mrRunningJobConfig.getEagleServiceConfig(), mrRunningJobConfig.getEndpointConfig(), mrRunningJobConfig.getZkStateConfig(), confKeyKeys, config), tasks).setNumTasks(tasks).fieldsGrouping(spoutName, APP_TO_JOB_STREAM, new Fields("appId")); topologyBuilder.setBolt("mrRunningJobMetricBolt", new MRRunningAppMetricBolt(mrRunningJobConfig), 1) .setNumTasks(1).shuffleGrouping(spoutName, APP_TO_METRIC_STREAM); topologyBuilder.setBolt("acceptedAppSink", environment.getStreamSink("ACCEPTED_APP_STREAM", config), 1) .setNumTasks(1).shuffleGrouping("mrRunningJobMetricBolt"); return topologyBuilder.createTopology(); }  @Override StormTopology execute(Config config, StormEnvironment environment);  }
@Test public void testMRRunningJobParseBolt() throws Exception { mockStatic(Executors.class); ExecutorService executorService = mock(ExecutorService.class); when(Executors.newFixedThreadPool(anyInt())).thenReturn(executorService); Config config = ConfigFactory.load(); MRRunningJobConfig mrRunningJobConfig = MRRunningJobConfig.newInstance(config); List<String> confKeyKeys = makeConfKeyKeys(mrRunningJobConfig); MRRunningJobParseBolt mrRunningJobParseBolt = new MRRunningJobParseBolt( mrRunningJobConfig.getEagleServiceConfig(), mrRunningJobConfig.getEndpointConfig(), mrRunningJobConfig.getZkStateConfig(), confKeyKeys, config); MRRunningJobManager mrRunningJobManager = mock(MRRunningJobManager.class); PowerMockito.whenNew(MRRunningJobManager.class).withArguments(mrRunningJobConfig.getZkStateConfig()).thenReturn(mrRunningJobManager); mrRunningJobParseBolt.prepare(null, null, null); InputStream previousmrrunningapp = this.getClass().getResourceAsStream("/previousmrrunningapp.json"); AppsWrapper appsWrapper = OBJ_MAPPER.readValue(previousmrrunningapp, AppsWrapper.class); List<AppInfo> appInfos = appsWrapper.getApps().getApp(); AppInfo app1 = appInfos.get(0); Tuple tuple = Testing.testTuple(new Values(app1.getId(), app1, null)); mrRunningJobParseBolt.execute(tuple); Field runningMRParsers = MRRunningJobParseBolt.class.getDeclaredField("runningMRParsers"); runningMRParsers.setAccessible(true); Map<String, MRJobParser> appIdToMRJobParser = (Map<String, MRJobParser>) runningMRParsers.get(mrRunningJobParseBolt); Assert.assertEquals(1, appIdToMRJobParser.size()); Assert.assertTrue(appIdToMRJobParser.get("application_1479206441898_30784") != null); Assert.assertTrue(appIdToMRJobParser.get("application_1479206441898_30784").status().equals(MRJobParser.ParserStatus.RUNNING)); verify(executorService, times(1)).execute(appIdToMRJobParser.get("application_1479206441898_30784")); verify(executorService, times(1)).execute(any(MRJobParser.class)); MRJobParser mrJobParser = appIdToMRJobParser.get("application_1479206441898_30784"); mrJobParser.setStatus(MRJobParser.ParserStatus.APP_FINISHED); AppInfo app2 = appInfos.get(1); tuple = Testing.testTuple(new Values(app2.getId(), app2, null)); mrRunningJobParseBolt.execute(tuple); Map<String, MRJobParser> appIdToMRJobParser1 = (Map<String, MRJobParser>) runningMRParsers.get(mrRunningJobParseBolt); Assert.assertEquals(1, appIdToMRJobParser1.size()); Assert.assertTrue(appIdToMRJobParser1.get("application_1479206441898_30784") == null); Assert.assertTrue(appIdToMRJobParser1.get("application_1479206441898_35341") != null); Assert.assertTrue(appIdToMRJobParser1.get("application_1479206441898_35341").status().equals(MRJobParser.ParserStatus.RUNNING)); verify(executorService, times(1)).execute(appIdToMRJobParser.get("application_1479206441898_35341")); verify(executorService, times(2)).execute(any(MRJobParser.class)); app2 = appInfos.get(1); tuple = Testing.testTuple(new Values(app2.getId(), app2, null)); mrRunningJobParseBolt.execute(tuple); Map<String, MRJobParser> appIdToMRJobParser2 = (Map<String, MRJobParser>) runningMRParsers.get(mrRunningJobParseBolt); Assert.assertEquals(1, appIdToMRJobParser2.size()); Assert.assertTrue(appIdToMRJobParser2.get("application_1479206441898_30784") == null); Assert.assertTrue(appIdToMRJobParser2.get("application_1479206441898_35341") != null); Assert.assertTrue(appIdToMRJobParser2.get("application_1479206441898_35341").status().equals(MRJobParser.ParserStatus.RUNNING)); verify(executorService, times(2)).execute(any(MRJobParser.class)); }