@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);
		}
	}
}