static PCollection<String> joinEvents(
      PCollection<TableRow> eventsTable, PCollection<TableRow> countryCodes) throws Exception {

    final TupleTag<String> eventInfoTag = new TupleTag<>();
    final TupleTag<String> countryInfoTag = new TupleTag<>();

    // transform both input collections to tuple collections, where the keys are country
    // codes in both cases.
    PCollection<KV<String, String>> eventInfo =
        eventsTable.apply(ParDo.of(new ExtractEventDataFn()));
    PCollection<KV<String, String>> countryInfo =
        countryCodes.apply(ParDo.of(new ExtractCountryInfoFn()));

    // country code 'key' -> CGBKR (<event info>, <country name>)
    PCollection<KV<String, CoGbkResult>> kvpCollection =
        KeyedPCollectionTuple.of(eventInfoTag, eventInfo)
            .and(countryInfoTag, countryInfo)
            .apply(CoGroupByKey.create());

    // Process the CoGbkResult elements generated by the CoGroupByKey transform.
    // country code 'key' -> string of <event info>, <country name>
    PCollection<KV<String, String>> finalResultCollection =
        kvpCollection.apply(
            "Process",
            ParDo.of(
                new DoFn<KV<String, CoGbkResult>, KV<String, String>>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                    KV<String, CoGbkResult> e = c.element();
                    String countryCode = e.getKey();
                    String countryName = e.getValue().getOnly(countryInfoTag);
                    for (String eventInfo : c.element().getValue().getAll(eventInfoTag)) {
                      // Generate a string that combines information from both collection values
                      c.output(
                          KV.of(
                              countryCode,
                              "Country name: " + countryName + ", Event info: " + eventInfo));
                    }
                  }
                }));

    // write to GCS
    PCollection<String> formattedResults =
        finalResultCollection.apply(
            "Format",
            ParDo.of(
                new DoFn<KV<String, String>, String>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) {
                    String outputstring =
                        "Country code: " + c.element().getKey() + ", " + c.element().getValue();
                    c.output(outputstring);
                  }
                }));
    return formattedResults;
  }
@Test
  @Category(ValidatesRunner.class)
  public void testJoin() throws java.lang.Exception {
    PCollection<TableRow> input1 = p.apply("CreateEvent", Create.of(EVENT_ARRAY));
    PCollection<TableRow> input2 = p.apply("CreateCC", Create.of(CC_ARRAY));

    PCollection<String> output = JoinExamples.joinEvents(input1, input2);
    PAssert.that(output).containsInAnyOrder(JOINED_EVENTS);
    p.run().waitUntilFinish();
  }