Add deduplication logic for kafka sink in case of spark task retries
Nessuno ha ancora preso questa issue.
Valutazione
- Difficoltà
- 5/5
- Tempo stimato
- Più di una settimana
- Idoneità per principianti
- 25/100
- Tipo di issue
- Funzionalità
- Chiarezza
- Abbastanza chiara
- Stato di attività
- Ferma
- Stack tecnologico
- kafka, scala
- Ambito
- stream-processing
Direzione di ricerca
Inizia esaminando il DeduplicateKafkaSinkTransformer esistente e il numero del tentativo del task Spark disponibile durante i retries. Valuta un Kafka ProducerInterceptor che gestisca i retries tra gli executor e mantenga una ricerca in memoria per i retries nello stesso executor; il lavoro è considerato completato quando i duplicati vengono scartati o reindirizzati a un trash topic secondo le ipotesi indicate.
Scritto dal modello di indicizzazione a partire dal testo della issue.
Descrizione
Problem description
Spark does not provide an exactly-once behaviour for the Kafka sink, but only at-least-once, and will probably never do so (https://github.com/apache/spark/pull/25618). Under certain assumptions (no concurrent producers, only 1 destination topic, not too big micro-batches, messages don't change between retries), idempotency can still be achieved. See #177.
The DeduplicateKafkaSinkTransformer only addresses retries on an application level. However, retries (and therefore duplicates) may happen on lower levels as well, namely:
- Retry of a
DataWritingSparkTask(the Spark task that will invoke the KafkaProducer) - Internal retry of the
KafkaProducer(when encountering aRetriableException, e.g. server disconnected)
Duplicates due to an internal retry of theKafkaProducercan be prevented by settingacks=allandenable.idempotenceon the kafka writer. However, this does not take into account retries of theDataWritingSparkTaskwhich invokes theKafkaProducer. For example, if there is an intermittentTopicAuthorizationException, theKafkaProducerwill fail and not retry, but theDataWritingSparkTaskwill retry nevertheless. In such a case, duplications are still possible. Another example is executor failure due to exceeding memory limits. If an executor exceeds memory limits during theDataWritingSparkTask, it will be terminated and another executor will retry the task, which may again lead to duplicates on the destination topic.
One solution is to switch off spark task retries, but obviously, this causes other problems.
Solution
It might be possible to implement a org.apache.kafka.clients.producer.ProducerInterceptor which would deduplicate retries messages in a similar way as the DeduplicateKafkaSinkTransformer. This would capture duplicates in a scenario where the retry happens on a different executor than the original executor (memory exceeded scenario). In addition, the interceptor should keep a lookup set in memory to capture duplicates that occurred due to retries on the same executor
(TopicAuthorizationException scenario).
A reattempt could be recognized through the attempt nr of the Spark task.
If messages cannot be dropped through the ProducerInterceptor, they could at least be redirected to a trash-topic
- Lingua principale
- Scala
- Stelle
- 47
- Fork
- 14
- Metriche di merge delle PR
- Nessuna PR unita negli ultimi 30g
Guida per i contributori
Nessuna guida per i contributori indicizzata per questo repository
Come iniziare
- Leggi tutta la issue e poi la guida ai contributi del progetto.
- Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
- Fai un fork del repository e lavora su un branch.
- Apri una pull request che faccia riferimento al numero della issue.
Altre issue di AbsaOSS/hyperdrive
-
Difficoltà 2/5 1-3 ore Idoneità per principianti 45/100
AbsaOSS/hyperdrive#269 ·
-
enhancement
Difficoltà 3/5 1-2 giorni Idoneità per principianti 42/100
AbsaOSS/hyperdrive#239 ·
-
bug
Difficoltà 3/5 1-2 giorni Idoneità per principianti 35/100
AbsaOSS/hyperdrive#230 ·
-
Atum integration Aperta
Difficoltà 5/5 Più di una settimana Idoneità per principianti 20/100
AbsaOSS/hyperdrive#211 · 1 commento · 1 reazione · 1 assegnatario ·
-
internal-task
Difficoltà 4/5 3-5 giorni Idoneità per principianti 28/100
AbsaOSS/hyperdrive#148 · 1 assegnatario ·
Tutte le issue di AbsaOSS/hyperdrive
Issue simili
-
RS_PixelAsPoint raises out of grid where RS_PixelAsCentroid and RS_PixelAsPolygon extrapolate Aperta
Difficoltà 2/5 1-3 ore Idoneità per principianti 75/100
-
Area: Excel support
Difficoltà 2/5 1-3 ore Idoneità per principianti 75/100
orbeon/orbeon-forms#7893 ·
-
x:action/fix x:module/practice-exercise x:size/small
Difficoltà 2/5 1-3 ore Idoneità per principianti 75/100
-
Difficoltà 2/5 1-3 ore Idoneità per principianti 75/100
lichess-org/lila#21788 · 2 commenti ·
-
area:Iceberg area:writer documentation requires-triage
Difficoltà 2/5 1-3 ore Idoneità per principianti 75/100
apache/datafusion-comet#6147 ·