### Question:
Set { public static <T> PCollection<T> intersection(PCollection<T> coll1, PCollection<T> coll2) { return Cogroup.cogroup(toTable(coll1), toTable(coll2)) .parallelDo(new DoFn<Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>>, T>() { @Override public void process(Pair<T, Pair<Collection<Boolean>, Collection<Boolean>>> input, Emitter<T> emitter) { Pair<Collection<Boolean>, Collection<Boolean>> groups = input.second(); if (!groups.first().isEmpty() && !groups.second().isEmpty()) { emitter.emit(input.first()); } } }, coll1.getPType()); }  static PCollection<T> difference(PCollection<T> coll1,
      PCollection<T> coll2); static PCollection<T> intersection(PCollection<T> coll1,
      PCollection<T> coll2); static PCollection<Tuple3<T, T, T>> comm(PCollection<T> coll1,
      PCollection<T> coll2);  }### Answer:
@Test public void testIntersection() throws Exception { PCollection<String> intersection = Set.intersection(set1, set2); assertEquals(Lists.newArrayList("a", "c"), Lists.newArrayList(intersection.materialize())); }