CoGroup extends InnerJoin<T> { public CoGroup() { } CoGroup(); @Override List<List<T>> getOutput(List<Set<T>> accumulatedValue);  }
@Test public void CoGroupTest() { CoGroup<Long> cg = new CoGroup<Long>(); List<Set<Long>> accu = cg.defaultAccumulatedValue(); Assert.assertEquals(2, accu.size()); for (int i = 0; i < 2; i++) { Assert.assertEquals(0, accu.get(i).size()); } for (long i = 1; i <= 3; i++) { accu = cg.accumulate(accu, i); accu = cg.accumulate2(accu, i * 2); } for (int i = 0; i < 2; i++) { Assert.assertEquals(3, accu.get(i).size()); } Assert.assertEquals(2, cg.getOutput(accu).size()); for (int i = 0; i < 2; i++) { Assert.assertEquals(3, cg.getOutput(accu).get(i).size()); } Assert.assertTrue(1 == cg.getOutput(accu).get(0).get(0)); Assert.assertTrue(4 == cg.getOutput(accu).get(1).get(1)); }