[Bug]: Race condition in Python SDK Harness ProcessBundleProgress

### What happened?

Hello, I am on Apache Beam v2.35.0 running on GCP Dataflow, and I've encountered what I believe are race conditions in the progress reporting machinery (i.e. `process_bundle_progress` or `ProcessBundleProgressRequest`):

```
Error processing instruction process_bundle_progress-1213241858972398550-1099. Original traceback is
Traceback (most recent call last):
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 267, in _execute
response = task()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 302, in
lambda: self.create_worker().do_instruction(request), request)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 581, in do_instruction
getattr(request, request_type), request.instruction_id)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 672, in process_bundle_progress
monitoring_infos = processor.monitoring_infos()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/bundle_processor.py", line 1131, in monitoring_infos
op.monitoring_infos(transform_id, dict(tag_to_pcollection_id)))
File "apache_beam/runners/worker/operations.py", line 356, in apache_beam.runners.worker.operations.Operation.monitoring_infos
File "apache_beam/runners/worker/operations.py", line 360, in apache_beam.runners.worker.operations.Operation.monitoring_infos
File "apache_beam/runners/worker/operations.py", line 425, in apache_beam.runners.worker.operations.Operation.execution_time_monitoring_infos
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/metrics/monitoring_infos.py", line 204, in int64_counter
return create_monitoring_info(urn, SUM_INT64_TYPE, metric, labels)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/metrics/monitoring_infos.py", line 303, in create_monitoring_info
urn=urn, type=type_urn, labels=labels or {}, payload=payload)
SystemError: returned NULL without setting an error
```

```
Error processing instruction process_bundle_progress-5696618351405637733-1593. Original traceback is
Traceback (most recent call last):
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 267, in _execute
response = task()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 302, in
lambda: self.create_worker().do_instruction(request), request)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 581, in do_instruction
getattr(request, request_type), request.instruction_id)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 672, in process_bundle_progress
monitoring_infos = processor.monitoring_infos()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/bundle_processor.py", line 1131, in monitoring_infos
op.monitoring_infos(transform_id, dict(tag_to_pcollection_id)))
File "apache_beam/runners/worker/operations.py", line 356, in apache_beam.runners.worker.operations.Operation.monitoring_infos
File "apache_beam/runners/worker/operations.py", line 360, in apache_beam.runners.worker.operations.Operation.monitoring_infos
File "apache_beam/runners/worker/operations.py", line 413, in apache_beam.runners.worker.operations.Operation.execution_time_monitoring_infos
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/metrics/monitoring_infos.py", line 204, in int64_counter
return create_monitoring_info(urn, SUM_INT64_TYPE, metric, labels)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/metrics/monitoring_infos.py", line 303, in create_monitoring_info
urn=urn, type=type_urn, labels=labels or {}, payload=payload)
File "/data_gen_docker_pybinary.runfiles/python3_x86_64/lib/python3.7/_collections_abc.py", line 840, in update
for key in other:
RuntimeError: dictionary changed size during iteration
```

```
Error processing instruction process_bundle_progress-6997054913682226470-568. Original traceback is
Traceback (most recent call last):
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 267, in _execute
response = task()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 302, in
lambda: self.create_worker().do_instruction(request), request)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 581, in do_instruction
getattr(request, request_type), request.instruction_id)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 683, in process_bundle_progress
for info in monitoring_infos
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 683, in
for info in monitoring_infos
AttributeError: 'bytes' object has no attribute 'payload'
```

```
Error processing instruction process_bundle_progress-5538519205655791026-1654. Original traceback is
Traceback (most recent call last):
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 267, in _execute
response = task()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 302, in
lambda: self.create_worker().do_instruction(request), request)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 581, in do_instruction
getattr(request, request_type), request.instruction_id)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/sdk_worker.py", line 672, in process_bundle_progress
monitoring_infos = processor.monitoring_infos()
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/runners/worker/bundle_processor.py", line 1131, in monitoring_infos
op.monitoring_infos(transform_id, dict(tag_to_pcollection_id)))
File "apache_beam/runners/worker/operations.py", line 356, in apache_beam.runners.worker.operations.Operation.monitoring_infos
File "apache_beam/runners/worker/operations.py", line 362, in apache_beam.runners.worker.operations.Operation.monitoring_infos
File "apache_beam/runners/worker/operations.py", line 779, in apache_beam.runners.worker.operations.DoOperation.pcollection_count_monitoring_infos
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/metrics/monitoring_infos.py", line 238, in int64_distribution
return create_monitoring_info(urn, DISTRIBUTION_INT64_TYPE, payload, labels)
File "/data_gen_docker_pybinary.runfiles/cruise_ws/cruise/mlp/cfs/projects/trajectory_ranking/pipeline/data_gen_docker_pybinary_exedir/apache_beam/metrics/monitoring_infos.py", line 303, in create_monitoring_info
urn=urn, type=type_urn, labels=labels or {}, payload=payload)
File "/data_gen_docker_pybinary.runfiles/python3_x86_64/lib/python3.7/_collections_abc.py", line 841, in update
self[key] = other[key]
ValueError: b'\x00\x00\x00\x00\xda?... REDACTED' has type str, but isn't valid UTF-8 encoding. Non-UTF-8 strings must be converted to unicode objects before being added.
```

I am running long running C++ code through pybind11 which I think might be a contributing factor. However my C++ code does not access any Python objects without holding the GIL and definitely doesn't change anything related to progress reporting.

I am marking this as P1 because I assume race conditions can cause data loss, etc - sorry if this is inappropriate and feel free to change it.

### Issue Priority

Priority: 1 (data loss / total loss of function)

### 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