[Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute

### What needs to happen?

The minimal working example below works (with `DirectRunner`) with Python 3.9 and Apache Beam 2.38.0 but fails on Apache Beam 2.39.0 and 2.44.0.

Using Apache Beam 2.39.0 and 2.44.0, the example fails with the error `AssertionError: A total of 2 watermark-pending bundles did not execute.`. When I switch the logging to `DEBUG`, I see messages of the form `Unable to add bundle for stage` along with `Stage input watermark: Timestamp(-9223372036854.775000)` (i.e. `timestamp.MIN_TIMESTAMP`) and `Bundle schedule watermark: Timestamp(9223372036854.775000)` (i.e. `timestamp.MAX_TIMESTAMP`) for the two bundles.

I have marked this issue as a task because I am not sure if this change in behaviour is by design in version 2.39.0 or if it is actually an issue / bug.

Note: I have opened a question [here](https://stackoverflow.com/q/75961407) on this also.

```py
import logging
import apache_beam as beam

def setup_logging():
log_format = '[%(asctime)-15s] [%(name)s] [%(levelname)s]: %(message)s'
logging.basicConfig(format=log_format, level=logging.INFO)
logging.info("Pipeline Started")

class CreateKvPCollectWithSideInputDoFn(beam.DoFn):
def __init__(self):
super().__init__()

def process(self, element, side_input):
print(f"side_input_type: {type(side_input)}")
yield "b", "2"

class CreateKvPCollectDoFn(beam.DoFn):
def __init__(self):
super().__init__()

def process(self, element):
yield "a", "1"

def main():
setup_logging()

pipeline = beam.Pipeline()

pcollect_input = (
pipeline
| "Input/Create" >> beam.Create(["input"])
)

kvpcollect_1 = (
pcollect_input | "PCollection_1" >> beam.ParDo(CreateKvPCollectDoFn())
)
beamdict_1 = beam.pvalue.AsDict(kvpcollect_1)

kvpcollect_2 = (
pcollect_input
| "PCollection_2" >> beam.ParDo(
CreateKvPCollectWithSideInputDoFn(), side_input=beamdict_1
)
# Commenting in this line allows the example to work on 2.39.0
# | "Reshuffle_2" >> beam.Reshuffle()
)

kvpcollect_3 = (
(kvpcollect_1, kvpcollect_2)
| "Flatten" >> beam.Flatten()
)
beamdict_3 = beam.pvalue.AsDict(kvpcollect_3)

(
pcollect_input
| "UseBeamDict_3" >> beam.ParDo(CreateKvPCollectWithSideInputDoFn(), side_input=beamdict_3)
| "PrintResult" >> beam.Map(print)
)

result = pipeline.run()
result.wait_until_finish()

if __name__ == '__main__':
main()

```

### Issue Priority

Priority: 2 (default / most normal work should be filed as P2)

### Issue Components

- [X] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner