GCP Dataflow Job scenarios with 'Input is Empty' While Processing Pub/Sub Messages from Multiple Topics
Could someone explain I'm confused about I'm writing unit tests and I'm building a feature where I'm reviewing some code and I've been banging my head against this for hours. I'm running a Dataflow pipeline to process messages from two Pub/Sub topics, but the job fails with the behavior 'Input is Empty' after a few minutes. The job is set to use the Apache Beam SDK version 2.33.0. The relevant part of my pipeline code is as follows: ```python import apache_beam as beam from apache_beam.io import ReadFromPubSub, WriteToBigQuery from apache_beam.options.pipeline_options import PipelineOptions options = PipelineOptions( runner='DataflowRunner', project='my-gcp-project', temp_location='gs://my-bucket/temp', region='us-central1' ) def process_message(message): # Process your message logic here return message with beam.Pipeline(options=options) as pipeline: messages = ( pipeline | 'Read from Topic 1' >> ReadFromPubSub(subscription='projects/my-gcp-project/subscriptions/my-subscription-1') | 'Read from Topic 2' >> ReadFromPubSub(subscription='projects/my-gcp-project/subscriptions/my-subscription-2') | 'Process Messages' >> beam.Map(process_message) | 'Write to BigQuery' >> WriteToBigQuery( table='my-dataset.my-table', schema='SCHEMA_AUTODETECT' ) ) ``` I've confirmed that messages are being published to both subscriptions. I can see them in the Pub/Sub dashboard, but the Dataflow job starts and then immediately fails, indicating that no input data is being processed. I have also tried increasing the `--max_num_workers` option to see if that helps, but the scenario continues. I'm using a region specific to my project and have checked the Cloud Pub/Sub IAM roles for my Dataflow service account to ensure it has the necessary permissions. Can someone guide to identify why my Dataflow job is reporting 'Input is Empty'? What could be the reason for this behavior when I'm trying to read from two Pub/Sub topics? Am I missing something obvious? This issue appeared after updating to Python 3.9. I'd really appreciate any guidance on this. This is my first time working with Python 3.10. This issue appeared after updating to Python 3.10. Has anyone dealt with something similar? Am I approaching this the right way? This is for a REST API running on CentOS. Thanks for taking the time to read this!