Ошибка потока данных при предоставлении темы pubsub в качестве документа

Я частично решил эту проблему, так как мой конвейер неправильно пытался опубликовать неудачные вставки в Big Query, однако у меня все еще есть проблема, заключающаяся в том, что я не могу передать имя темы pubsub в качестве входного параметра. Однако это работает, если название темы жестко запрограммировано.

#################################################################
# Import the libraries required by the pipeline                 #
#################################################################
from __future__ import absolute_import

import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.options.pipeline_options import StandardOptions
from apache_beam.options.value_provider import RuntimeValueProvider
import json
import time
from datetime import datetime
import dateutil.parser
import sys
import logging

#################################################################
# Create a class for the user defined settings provided at job 
# creation
#################################################################

class userOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            '--subscription',  
            default='projects/MYPROJECT/subscrpiptions/subscription', 
            help='PubSub subscription to listen on')
        parser.add_value_provider_argument(
            '--bqtable', 
            default='dataset.table', 
            help='Big Query Table Name in the format project:dataset.table') 
        parser.add_value_provider_argument(
            '--topic', 
            default='projects/MYPROJECT/topics/subscription', 
            help='Pubsub topic to write failures to') 

##############################################################################
# Format failure message
##############################################################################
def format_failed_message(data):
    try:
        message=json.dumps(data)
    except:
        print("customError in function format_failed_message occured.", sys.exc_info(), "Message contents: ", data)
    return message

#################################################################
# create a function called run                                  #
#################################################################
def run():

    ##############################################################
    # Setup the pipeline options with both passed in arguments 
    # and streaming options
    ##############################################################
    user_options = PipelineOptions().view_as(userOptions)

    pipeline_options = PipelineOptions()
    pipeline_options.view_as(SetupOptions).save_main_session = True
    pipeline_options.view_as(StandardOptions).streaming = True

    ##############################################################
    # Define the pipeline
    ##############################################################
    with beam.Pipeline(options=pipeline_options) as p:

        # First we create a PCollection which will contain the messages read from Pubsub
        records = ( 
        p  | 'Read from PubSub' 
            >> beam.io.ReadFromPubSub(
                subscription=str(user_options.subscription),
                id_label='Message_ID',
                with_attributes=True)
        # Transform the message and its attributes to a dict.
        | 'Format Message' >> 
            beam.Map(format_message_element)
        # Transform the empty arrays defined as element:null to element:[].
        | 'Transform null records to empty list' >>
            beam.Map(transform_null_records)
        # Transform the dateCreated and DateModified to a big query compatible timestamp format.
        | 'Transform Dates' >>
            beam.Map(format_dates)
        # Attempt to write the rows to BQ
        | 'Write to Big Query' >>
            beam.io.gcp.bigquery.WriteToBigQuery(
                table=user_options.bqtable,
                create_disposition='CREATE_IF_NEEDED',
                write_disposition='WRITE_APPEND',
                insert_retry_strategy='RETRY_NEVER'
            )
        )

        #For any rows that failed to write to BQ
        failed_data = (records[beam.io.gcp.bigquery.BigQueryWriteFn.FAILED_ROWS]
                        #Format the dictionary to a string
                        | 'Format the dictionary as a string for publishing' >>
                            beam.Map(format_failed_message)
                        #Encode the string to utf8 bytes
                        | 'Encode the message' >>
                            beam.Map(lambda x: x.encode('utf-8')).with_output_types(bytes)
                        )
        #Published the failed rows to pubsub
        failed_data | beam.io.gcp.pubsub.WriteToPubSub(topic='projects/xxxx-xxxxx-xxxxxx/topics/top-xxxxx-failures')
        #failed_data | beam.io.gcp.pubsub.WriteToPubSub(topic=user_options.topic)

    # As this is a streaming pipeline it will run continuosly till either we 
    # stop the pipeline or it fails.
    result = p.run()
    result.wait_until_finish()

#At the main entry point call the run function
if __name__ == '__main__':
    #logging.getLogger().setLevel(logging.INFO)
    run()