[Bug]: Python Nexmark Query 4,6,9 is broken 'bytes' object has no attribute 'end'

### What happened?
This affects 4, 6, and 9. CLI below is for 4.

```
./gradlew --continue -Dorg.gradle.jvmargs=-Xms2g -Dorg.gradle.jvmargs=-Xmx6g -Dorg.gradle.vfs.watch=false -Pdocker-pull-licenses -Pnexmark.runner=:runners:direct-java "-Pnexmark.args=--manageResources=false --monitorJobs=true --bigQueryTable=nexmark --bigQueryDataset=nexmark_PRs --project=apache-beam-testing --resourceNameMode=QUERY_RUNNER_AND_MODE --exportSummaryToBigQuery=true --tempLocation=gs://temp-storage-for-perf-tests/nexmark --influxDatabase=beam_test_metrics --influxHost=http://10.128.0.96:8086/ --baseInfluxMeasurement=nexmark_PRs --exportSummaryToInfluxDB=true --influxRetentionPolicy=forever --suite=SMOKE --enforceEncodability=true --enforceImmutability=true --runner=DirectRunner --query=4 --numEvents=10000 --generateEventFilePathPrefix=gs://temp-storage-for-perf-tests/nexmark/eventFiles/jenkins-beam_PostCommit_Python_Nexmark_Direct_PR-6/query4-" :sdks:java:testing:nexmark:run

./gradlew --continue -Dorg.gradle.jvmargs=-Xms2g -Dorg.gradle.jvmargs=-Xmx6g -Dorg.gradle.vfs.watch=false -Pdocker-pull-licenses "-Pnexmark.args= --monitor_jobs --big_query_table=nexmark --big_query_dataset=nexmark_PRs --project=apache-beam-testing --resource_name_mode=QUERY_RUNNER_AND_MODE --export_summary_to_big_query --temp_location=gs://temp-storage-for-perf-tests/nexmark --influx_database=beam_test_metrics --influx_host=http://10.128.0.96:8086/ --base_influx_measurement=nexmark_PRs --export_summary_to_influx_db --influx_retention_policy=forever --suite=SMOKE --enforce_encodability --enforce_immutability --runner=DirectRunner --query=4 --num_events=10000 --input=gs://temp-storage-for-perf-tests/nexmark/eventFiles/jenkins-beam_PostCommit_Python_Nexmark_Direct_PR-6/query4-\*" :sdks:python:apache_beam:testing:benchmarks:nexmark:run
```

for environment ref_Environment_default_environment_1 (beam:env:embedded_python:v1, b'')
12:32:25 (MainThread) Finished listing 1 files in 0.04667186737060547 seconds.
12:32:25 Traceback (most recent call last):
12:32:29 File "/usr/lib/python3.8/runpy.py", line 194, in _run_module_as_main
12:32:29 return _run_code(code, main_globals, None,
12:32:29 File "/usr/lib/python3.8/runpy.py", line 87, in _run_code
12:32:29 exec(code, run_globals)
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/nexmark_launcher.py", line 528, in
12:32:29 launcher.run()
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/nexmark_launcher.py", line 513, in run
12:32:29 self.run_query(
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/nexmark_launcher.py", line 306, in run_query
12:32:29 result = self.pipeline.run()
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/pipeline.py", line 577, in run
12:32:29 return self.runner.run_pipeline(self, self._options)
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/direct/direct_runner.py", line 131, in run_pipeline
12:32:29 return runner.run_pipeline(pipeline, options)
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 201, in run_pipeline
12:32:29 self._latest_run_result = self.run_via_runner_api(
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 222, in run_via_runner_api
12:32:29 return self.run_stages(stage_context, stages)
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 453, in run_stages
12:32:29 bundle_results = self._execute_bundle(
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 781, in _execute_bundle
12:32:29 self._run_bundle(
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1010, in _run_bundle
12:32:29 result, splits = bundle_manager.process_bundle(
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1337, in process_bundle
12:32:29 self._send_input_to_****(process_bundle_id, transform_id, elements)
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1211, in _send_input_to_****
12:32:29 for byte_stream in (byte_streams or []):
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py", line 308, in __iter__
12:32:29 return itertools.chain(*self.partition(1))
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py", line 294, in partition
12:32:29 for wkvs in windowed_key_values(key, windowed_values):
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/transforms/trigger.py", line 1232, in process_entire_key
12:32:29 for wvalue in self.process_elements(state,
12:32:29 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/transforms/trigger.py", line 1384, in process_elements
12:32:29 if input_watermark > window.end + self.allowed_lateness:
12:32:29 AttributeError: 'bytes' object has no attribute 'end'
12:32:29

### Issue Priority

Priority: 3 (minor)

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