[Bug]: Unable to create exactly-once Flink pipeline with stream source and file sink

### What happened?

Beam 2.46.0
Java 11
Flink 1.15

My pipeline is consuming from streaming source (AWS Kinesis) and writes to parquet files once per minute.

When restarting it from a Flink snapshot, I was unable to get exactly-once delivery of the records into the final files.

1. When using standard file-naming policy, I get files like `out-2023-03-29T22-45-00.000Z-2023-03-29T22-46-00.000Z-00000-of-00004.parquet`. I've noticed that, after a restart, some of the window files are replaced, which causes records to be lost.
2. I've created a JobId var which is re-generated every time the app starts, which gives file names like `out--2023-03-29T22-45-00.000Z-2023-03-29T22-46-00.000Z-00000-of-00004.parquet` - this allows me to avoid having replaced files and lost records, but introduces duplicates. The file names are made via https://github.com/psolomin/beam-playground/blob/e0f1834d6ee76a796d6d21836e1d04140e6d9ca2/kinesis-io-with-enhanced-fan-out/src/main/java/com/psolomin/consumer/NoColonFileNaming.java#L16

(2) effectively makes the pipeline to be at-least-once.

Also, when the app fails and Flink restarts it, I don't see neither duplicates, nor lost records, and exactly-once semantic actually works as expected.

I've tried different setting of the window:

```
Window.into(FixedWindows.of(Duration.standardSeconds(60)))
```

```
Window.into(FixedWindows.of(Duration.standardSeconds(60)))
.withAllowedLateness(Duration.ZERO) <- changing this to greater numbers had no effect
.discardingFiredPanes()
.triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(Duration.standardSeconds(60))))
```

Is it expected behaviour? Is it documented anywhere? Is there a configuration combo of savepoint config and window config which preserves exaclty-once when the app is stopped and started from a savepoint?

### Issue Priority

Priority: 3 (minor)

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

Beginnen Sie mit der im Issue beschriebenen Beam 2.46.0-Java-Pipeline und dem verlinkten Beispiel NoColonFileNaming.java. Reproduzieren Sie anschließend einen Flink-Savepoint-Neustart mit der standardmäßigen und der JobId-basierten Dateibenennung. Vergleichen Sie ersetzte Dateien, verlorene Datensätze und Duplikate; abgeschlossen ist die Arbeit, wenn geklärt ist, ob das Verhalten erwartungsgemäß ist, und die für eine Exactly-once-Zustellung erforderliche Savepoint- und Window-Konfiguration dokumentiert oder korrigiert wurde.

Vom Indexierungsmodell aus dem Issue-Text verfasst.