metrics-build-topology.java

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("word", new TestWordSpout(), 10);
builder.setBolt("exclaim1", new ExclamationBolt(), 3)
  .shuffleGrouping("word");
builder.setBolt("exclaim2", new ExclamationBolt(), 2)
  .shuffleGrouping("exclaim1");

Config conf = new Config();
conf.setDebug(true);
conf.registerMetricsConsumer(LoggingMetricsConsumer.class, 2);

if (args != null && args.length > 0) {
  conf.setNumWorkers(3);
  StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
}
else {
  LocalCluster cluster = new LocalCluster();
  cluster.submitTopology("test", conf, builder.createTopology());
  Utils.sleep(5*60*1000L);
  cluster.killTopology("test");
  cluster.shutdown();
}