@Test
public void testAggregateBy() throws Exception {

  Pipe p = Pump.prime()
      .groupby("line")
      .aggregateby(new MaxFunctor(), new Max(), "offset")
      .retain("max")
      .coerce("max", int.class)
      .toPipe();
  CascadingHelper.get().getFlowConnector().connect(Tests.getInTap(), Tests.getOutTap(), p).complete();
  assertEquals(Arrays.asList("10", "12", "22"), getOutputStrings());
}