### Question:
BoltCollector extends AbstractStormCollector<OUT> implements IOutputCollector { @Override public List<Integer> emit(final String streamId, final Collection<Tuple> anchors, final List<Object> tuple) { return this.tansformAndEmit(streamId, tuple); } BoltCollector(final HashMap<String, Integer> numberOfAttributes, final int taskId,
			final Collector<OUT> flinkOutput); @Override void reportError(final Throwable error); @Override List<Integer> emit(final String streamId, final Collection<Tuple> anchors, final List<Object> tuple); @Override void emitDirect(final int taskId, final String streamId, final Collection<Tuple> anchors, final List<Object> tuple); @Override void ack(final Tuple input); @Override void fail(final Tuple input);  }### Answer:
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testBoltStormCollector() throws InstantiationException, IllegalAccessException { for (int numberOfAttributes = -1; numberOfAttributes < 26; ++numberOfAttributes) { final Output flinkCollector = mock(Output.class); Tuple flinkTuple = null; final Values tuple = new Values(); BoltCollector<?> collector; final String streamId = "streamId"; HashMap<String, Integer> attributes = new HashMap<String, Integer>(); attributes.put(streamId, numberOfAttributes); if (numberOfAttributes == -1) { collector = new BoltCollector(attributes, -1, flinkCollector); tuple.add(new Integer(this.r.nextInt())); } else { collector = new BoltCollector(attributes, -1, flinkCollector); flinkTuple = Tuple.getTupleClass(numberOfAttributes).newInstance(); for (int i = 0; i < numberOfAttributes; ++i) { tuple.add(new Integer(this.r.nextInt())); flinkTuple.setField(tuple.get(i), i); } } final Collection anchors = mock(Collection.class); final List<Integer> taskIds; taskIds = collector.emit(streamId, anchors, tuple); Assert.assertNull(taskIds); if (numberOfAttributes == -1) { verify(flinkCollector).collect(tuple.get(0)); } else { verify(flinkCollector).collect(flinkTuple); } } }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testBoltStormCollectorWithTaskId() throws InstantiationException, IllegalAccessException { for (int numberOfAttributes = 0; numberOfAttributes < 25; ++numberOfAttributes) { final Output flinkCollector = mock(Output.class); final int taskId = 42; final String streamId = "streamId"; HashMap<String, Integer> attributes = new HashMap<String, Integer>(); attributes.put(streamId, numberOfAttributes); BoltCollector<?> collector = new BoltCollector(attributes, taskId, flinkCollector); final Values tuple = new Values(); final Tuple flinkTuple = Tuple.getTupleClass(numberOfAttributes + 1).newInstance(); for (int i = 0; i < numberOfAttributes; ++i) { tuple.add(new Integer(this.r.nextInt())); flinkTuple.setField(tuple.get(i), i); } flinkTuple.setField(taskId, numberOfAttributes); final Collection anchors = mock(Collection.class); final List<Integer> taskIds; taskIds = collector.emit(streamId, anchors, tuple); Assert.assertNull(taskIds); verify(flinkCollector).collect(flinkTuple); } }