Expand extends BaseFunction { @Override public void execute(TridentTuple tuple, TridentCollector collector) { Iterable iterable = (Iterable) tuple.getValue(0); Iterator iterator = iterable.iterator(); while (iterator.hasNext()) { collector.emit(new Values(iterator.next())); } }  @Override void execute(TridentTuple tuple, TridentCollector collector);  }
@Test (dataProvider = "getIterable") public void testExpand(Iterable iterable, int size) { final TridentTuple tuple = mock(TridentTuple.class); final TridentCollector collector = mock(TridentCollector.class); when(tuple.getValue(0)).thenReturn(iterable); Expand expand = new Expand(); expand.execute(tuple, collector); verify(tuple, times(1)).getValue(0); verify(collector, times(size)).emit(any(Values.class)); }