@@ -0,0 +1,2996 @@
+/**%0A * Licensed to the Apache Software Foundation (ASF) under one%0A * or more contributor license agreements.  See the NOTICE file%0A * distributed with this work for additional information%0A * regarding copyright ownership.  The ASF licenses this file%0A * to you under the Apache License, Version 2.0 (the%0A * %22License%22); you may not use this file except in compliance%0A * with the License.  You may obtain a copy of the License at%0A *%0A * http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage storm.starter;%0A%0Aimport backtype.storm.Config;%0Aimport backtype.storm.LocalCluster;%0Aimport backtype.storm.StormSubmitter;%0Aimport backtype.storm.metric.LoggingMetricsConsumer;%0Aimport backtype.storm.task.OutputCollector;%0Aimport backtype.storm.task.TopologyContext;%0Aimport backtype.storm.testing.TestWordSpout;%0Aimport backtype.storm.topology.OutputFieldsDeclarer;%0Aimport backtype.storm.topology.TopologyBuilder;%0Aimport backtype.storm.topology.base.BaseRichBolt;%0Aimport backtype.storm.tuple.Fields;%0Aimport backtype.storm.tuple.Tuple;%0Aimport backtype.storm.tuple.Values;%0Aimport backtype.storm.utils.Utils;%0A%0Aimport java.util.Map;%0A%0A/**%0A * This is a basic example of a Storm topology with metrics logging to the file.%0A */%0Apublic class ExclamationTopologyWithMetrics %7B%0A%0A  public static class ExclamationBolt extends BaseRichBolt %7B%0A    OutputCollector _collector;%0A%0A    @Override%0A    public void prepare(Map conf, TopologyContext context, OutputCollector collector) %7B%0A      _collector = collector;%0A    %7D%0A%0A    @Override%0A    public void execute(Tuple tuple) %7B%0A      _collector.emit(tuple, new Values(tuple.getString(0) + %22!!!%22));%0A      _collector.ack(tuple);%0A    %7D%0A%0A    @Override%0A    public void declareOutputFields(OutputFieldsDeclarer declarer) %7B%0A      declarer.declare(new Fields(%22word%22));%0A    %7D%0A%0A%0A  %7D%0A%0A  public static void main(String%5B%5D args) throws Exception %7B%0A    TopologyBuilder builder = new TopologyBuilder();%0A%0A    builder.setSpout(%22word%22, new TestWordSpout(), 10);%0A    builder.setBolt(%22exclaim1%22, new ExclamationBolt(), 3).shuffleGrouping(%22word%22).setNumTasks(6);%0A    builder.setBolt(%22exclaim2%22, new ExclamationBolt(), 2).shuffleGrouping(%22exclaim1%22).setNumTasks(6);%0A%0A    Config conf = new Config();%0A    conf.setDebug(true);%0A    conf.registerMetricsConsumer(LoggingMetricsConsumer.class);%0A%0A    if (args != null && args.length %3E 0) %7B%0A      conf.setNumWorkers(3);%0A      StormSubmitter.submitTopologyWithProgressBar(args%5B0%5D, conf, builder.createTopology());%0A    %7D%0A    else %7B%0A%0A      LocalCluster cluster = new LocalCluster();%0A      cluster.submitTopology(%22test%22, conf, builder.createTopology());%0A      Utils.sleep(10000);%0A      cluster.killTopology(%22test%22);%0A      cluster.shutdown();%0A    %7D%0A  %7D%0A%7D%0A

Add ExclamationTopologyWithMetrics to storm-starter