public static Impl join(By clause) {
    return new Impl(new JoinArguments(clause));
  }
@Test
  @Category(NeedsRunner.class)
  public void testUnderspecifiedCoGroup() {
    PCollection<Row> pc1 =
        pipeline
            .apply(
                "Create1",
                Create.of(Row.withSchema(CG_SCHEMA_1).addValues("user1", 1, "us").build()))
            .setRowSchema(CG_SCHEMA_1);
    PCollection<Row> pc2 =
        pipeline
            .apply(
                "Create2",
                Create.of(Row.withSchema(CG_SCHEMA_2).addValues("user1", 9, "us").build()))
            .setRowSchema(CG_SCHEMA_2);
    PCollection<Row> pc3 =
        pipeline.apply(
            "Create3", Create.of(Row.withSchema(CG_SCHEMA_3).addValues("user1", 17, "us").build()));

    thrown.expect(IllegalArgumentException.class);
    PCollectionTuple.of("pc1", pc1, "pc2", pc2, "pc3", pc3)
        .apply(
            "CoGroup",
            CoGroup.join("pc1", By.fieldNames("user", "country"))
                .join("pc2", By.fieldNames("user2", "country2")));
    pipeline.run();
  }