### Question:
KettleControlBolt extends BaseRichBolt { @SuppressWarnings("rawtypes") @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; componentToPendingTasks = new HashMap<String, List<Integer>>(); for (String componentId : leafSteps) { List<Integer> tasks = context.getComponentTasks(componentId); if (tasks == null || tasks.isEmpty()) { throw new IllegalStateException("No tasks defined for leaf step " + componentId); } componentToPendingTasks.put(componentId, new ArrayList<Integer>(tasks)); } notifier.init(stormConf); } KettleControlBolt(String transformationName, Notifier notifier,
      Set<String> leafSteps); @SuppressWarnings("rawtypes") @Override void prepare(Map stormConf, TopologyContext context,
      OutputCollector collector); @Override void execute(Tuple input); @Override void declareOutputFields(OutputFieldsDeclarer declarer); @Override void cleanup();  }### Answer:
@Test(expected = IllegalStateException.class) public void prepare_no_tasks_for_leaf_step_null() { KettleControlBolt bolt = new KettleControlBolt(TRANS_NAME, notifier, Collections.singleton(STEP_1)); EasyMock.expect(context.getComponentTasks(STEP_1)).andReturn(null); control.replay(); bolt.prepare(Collections.emptyMap(), context, collector); }
@Test(expected = IllegalStateException.class) public void prepare_no_tasks_for_leaf_step_empty() { KettleControlBolt bolt = new KettleControlBolt(TRANS_NAME, notifier, Collections.singleton(STEP_1)); EasyMock.expect(context.getComponentTasks(STEP_1)).andReturn( Collections.<Integer> emptyList()); control.replay(); bolt.prepare(Collections.emptyMap(), context, collector); }