pipline with tags, process error

PipelineWithTagForError.java

EventsProcessingOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                    .as(EventsProcessingOptions.class);

Pipeline p = Pipeline.create(options);

PCollectionTuple tableRows =
                  // read kafka topic
                   p.apply("kafka-topic-read", kafkaReader)
                    .apply("kafka-values", MapElements.into(TypeDescriptors.strings())
                            .via(record ->record.getKV().getValue()))
                    // convert value to JsonNode
                    .apply("string-to-json", ParseJsons.of(JsonNode.class))
                    // create TableRow
                    .apply("Build-table-row", ParDo.of(new EventsRowFn()).withOutputTags(MAIN_OUT, TupleTagList.of(DEADLETTER_OUT)));

                    
                  // save the MAIN tag to BQ
                  tableRows
                      .get(MAIN_OUT)
                      .apply("BQ-write", BigQueryIO.<TableRowWithEvent>write()
                      .to(tableSpec)
                      .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                      .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND);
                  
                  // save the DEADLETTER_OUT to BQ error table
                  tableRows
                    .get(DEADLETTER_OUT)
                    .apply("BQ-process-error-extract", ParDo.of(new BigQueryProcessErrorExtracFn()))
                    .apply("BQ-process-error-write", BigQueryIO.writeTableRows()
                            .to(errTableSpec)
                            .withJsonSchema(errSchema)
                            .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                            .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
  p.run();