docs: 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 = "none";
              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;
    }