[Bug]: KinesisIO processing-time watermarking can cause data loss

### What happened?

Beam 2.46.0
Java 11
Flink 1.15

According to the [doc](https://beam.apache.org/documentation/runners/flink/#pipeline-options-for-the-flink-runner) FlinkRunner supports options `parallelism` and `maxParallelism`. When `parallelism` is reduced between restarts with a savepoint, some bundles are never dispatched.

This also might be an issue related to window or sink de-serialization when the app is re-started from serialized state.

How to reproduce:
1. Start a Beam-Flink streaming job with `parallelism=3`, and `maxParallelism` not set. The job in the example had a window and file sink.
2. Stop the job with a savepoint
3. Re-start the job from the savepoint

If the job is then re-started again from that savepoint with original `parallelism=3`, the bundles are dispatched and pass through.

Setting `maxParallelism=3` and keeping it across app restarts does not change this behaviour. This behaviour does not occur when `parallelism` is increased, e.g. setting `parallelism` from `1` to `3` between savepoint-ed restarts makes all the data to sink.

This is a simplified job code (AWS Kinesis source, 3 shards):

```
PCollection windowedRecords = p.apply("Source", reader)
.apply("Fixed windows", Window.into(FixedWindows.of(Duration.standardSeconds(60))))
...
.apply(
"Sink to S3",
FileIO.write()
.via(ParquetIO.sink(ConsumedEvent.SCHEMA$)
.withCompressionCodec(CompressionCodecName.SNAPPY))
.to(opts.getSinkLocation()))
```

Toy project: https://github.com/psolomin/beam-playground/tree/parallelism-issue/kinesis-io-with-enhanced-fan-out#vanilla-flink

Full code snippet [class](https://github.com/psolomin/beam-playground/blob/parallelism-issue/kinesis-io-with-enhanced-fan-out/src/main/java/com/psolomin/consumer/KinesisToFilePipeline.java)

### Issue Priority

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

### Issue Components

- [ ] Component: Python SDK
- [X] 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
- [X] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner