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