import org.apache.storm.ILocalCluster; //导入依赖的package包/类
private TestJob createPirkTestJob(final Config config)
{
  final SpoutConfig kafkaConfig = setUpTestKafkaSpout(config);
  return new TestJob()
  {
    StormTopology topology = PirkTopology.getPirkTopology(kafkaConfig);

    @Override
    public void run(ILocalCluster iLocalCluster) throws Exception
    {
      iLocalCluster.submitTopology("pirk_integration_test", config, topology);
      logger.info("Pausing for setup.");
      // Thread.sleep(4000);
      // KafkaProducer producer = new KafkaProducer<String,String>(createKafkaProducerConfig());
      // loadTestData(producer);
      // Thread.sleep(10000);
      while (OutputBolt.latch.getCount() == testCountDown)
      {
        Thread.sleep(1000);
      }
      testCountDown -= 1;

      logger.info("Finished...");
    }
  };
}
 

import org.apache.storm.ILocalCluster; //导入依赖的package包/类
@Test
public void verifyEmittedValues() {
	MkClusterParam clusterParam = new MkClusterParam();
	clusterParam.setSupervisors(1);

	withSimulatedTimeLocalCluster(clusterParam, new TestJob() {
		
		@Override
		public void run(ILocalCluster cluster) throws JsonProcessingException {
			
			MockedSources mockedSources = new MockedSources();
			mockedSources.addMockData(builder.getSpoutId(), new Values(SWITCH_ID));

			Config config = new Config();
			config.setDebug(true);

			CompleteTopologyParam topologyParam = new CompleteTopologyParam();
			topologyParam.setMockedSources(mockedSources);
			topologyParam.setStormConf(config);

			Map<?, ?> result = completeTopology(cluster, builder.build(), topologyParam);
			assertTrue(multiseteq(new Values(new Values(SWITCH_ID)),
					readTuples(result, builder.getSpoutId())));
			assertTrue(multiseteq(new Values(new Values(SWITCH_ID)),
					readTuples(result, builder.getConfirmationBoltId())));
		}
		
	});
}
 

import org.apache.storm.ILocalCluster; //导入依赖的package包/类
public static Config submitTopology(ILocalCluster stormCluster, String topologyName,
                                    StormTopology stormTopology) throws Exception {
    Config stormConf = new Config();
    stormConf.putAll(Utils.readDefaultConfig());
    stormConf.put("storm.cluster.mode", "local");
    stormConf.setDebug(true);
    stormConf.setMaxTaskParallelism(3);
    stormConf.put(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN,
            org.apache.atlas.storm.hook.StormAtlasHook.class.getName());

    stormCluster.submitTopology(topologyName, stormConf, stormTopology);

    Thread.sleep(10000);
    return stormConf;
}
 

import org.apache.storm.ILocalCluster; //导入依赖的package包/类
public void testNormalizationTopology(NormalizationProcessor normalizationProcessor) throws Exception {

        final Config config = new Config();
        config.setDebug(true);
        final String topologyName = "SplitJoinTopologyTest";
        final StormTopology topology = createTopology(normalizationProcessor);
        log.info("Created topology with name: [{}] and topology: [{}]", topologyName, topology);

        ILocalCluster localCluster = new LocalCluster();
        log.info("Submitting topology: [{}]", topologyName);
        localCluster.submitTopology(topologyName, config, topology);
        Thread.sleep(2000);
        localCluster.shutdown();
    }
 

import org.apache.storm.ILocalCluster; //导入依赖的package包/类
protected void submitTopology() throws Exception {
    final Config config = getConfig();
    final String topologyName = "SplitJoinTopologyTest";
    final StormTopology topology = createTopology();
    log.info("Created topology with name: [{}] and topology: [{}]", topologyName, topology);

    ILocalCluster localCluster = new LocalCluster();
    log.info("Submitting topology: [{}]", topologyName);
    localCluster.submitTopology(topologyName, config, topology);

}
 

import org.apache.storm.ILocalCluster; //导入依赖的package包/类
public static ILocalCluster createLocalStormCluster() {
    // start a local storm cluster
    HashMap<String,Object> localClusterConf = new HashMap<>();
    localClusterConf.put("nimbus-daemon", true);
    return Testing.getLocalCluster(localClusterConf);
}
 

import org.apache.storm.ILocalCluster; //导入依赖的package包/类
protected void submitTopology() throws AlreadyAliveException, InvalidTopologyException {
    final Config config = getConfig();
    final String topologyName = "RulesTopologyTest";
    ILocalCluster localCluster = new LocalCluster();
    localCluster.submitTopology(topologyName, config, createTopology());
}