@@ -0,0 +1,1259 @@
+package com.example.storm;%0A%0Aimport backtype.storm.task.OutputCollector;%0Aimport backtype.storm.task.TopologyContext;%0Aimport backtype.storm.topology.IRichBolt;%0Aimport backtype.storm.topology.OutputFieldsDeclarer;%0Aimport backtype.storm.tuple.Fields;%0Aimport backtype.storm.tuple.Tuple;%0Aimport com.example.Marker;%0Aimport java.util.Arrays;%0Aimport java.util.Map;%0A%0A%0Apublic class MarkerBolt implements IRichBolt %7B%0A%0A%09private Marker marker;%0A%09private OutputCollector collector;%0A%0A%0A%09@Override%0A%09public void prepare(Map stormConf, TopologyContext topologyContext, OutputCollector outputCollector) %7B%0A%09%09collector = outputCollector;%0A%09%09marker = new Marker();%0A%09%7D%0A%0A%09@Override%0A%09public void execute(Tuple tuple) %7B%0A%09%09Long number = tuple.getLongByField(%22number%22);%0A%09%09String heading = tuple.getStringByField(%22heading%22);%0A%09%09Marker.Attributes bean = marker.mark(number);%0A%09%09collector.emit(tuple, Arrays.%3CObject%3EasList(heading, bean.getIsEven(), bean.getSource()));%0A%09%09collector.ack(tuple);%0A%09%7D%0A%0A%09@Override%0A%09public void cleanup() %7B%0A%09%7D%0A%0A%09@Override%0A%09public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) %7B%0A%09%09outputFieldsDeclarer.declare(new Fields(%22heading%22, %22isEven%22, %22source%22));%0A%09%7D%0A%0A%09@Override%0A%09public Map%3CString, Object%3E getComponentConfiguration() %7B%0A%09%09return null;%0A%09%7D%0A%0A%7D%0A

Simplify demo + bean output fields.