[Bug]: Unable to create exactly-once Flink pipeline with stream source and file sink
- Lingua principale
- Java
- Stelle
- 8.7k
- Fork
- 4.7k
- Merge medio
- 2g 9h
- PR unite (30g)
- 202
Descrizione
### 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
Guida per i contributori
Apri la guida per i contributori
Direzione di ricerca
Inizia con la pipeline Java Beam 2.46.0 descritta nell’issue e con l’esempio collegato NoColonFileNaming.java, quindi riproduci un riavvio da savepoint Flink con la denominazione dei file standard e basata su JobId. Confronta i file sostituiti, i record persi e i duplicati; il lavoro è completato quando viene stabilito se il comportamento è previsto e viene documentata o corretta la configurazione del savepoint e della finestra necessaria per una consegna exactly-once.
Scritto dal modello di indicizzazione a partire dal testo della issue.
Valutazione
- Stack tecnologico
- aws, java
- Ambito
- distributed-systems, stream-processing
- Tipo di issue
- Bug
- Difficoltà
- 4/5
- Tempo stimato
- 3-5 giorni
- Stato di attività
- Ferma
- Chiarezza
- Da chiarire
- Idoneità per principianti
- 25/100