Collect extends DataSetAnalyticBase<T, List<T>> { @Override public Collect<T> run(DataSet<T> input) throws Exception { super.run(input); serializer = input.getType().createSerializer(env.getConfig()); collectHelper = new CollectHelper<>(serializer); input .output(collectHelper) .name("Collect"); return this; }  @Override Collect<T> run(DataSet<T> input); @Override List<T> getResult();  }
@Test public void testEmptyList() throws Exception { DataSet<Long> dataset = env.fromCollection(Collections.emptyList(), TypeInformation.of(new TypeHint<Long>(){})); List<Long> collected = new Collect<Long>().run(dataset).execute(); assertEquals(0, collected.size()); }