Transactional producer's sender loops on AttributeError after a fatal error (2.2 and 2.3 branches)
Nessuno ha ancora preso questa issue.
Valutazione
- Difficoltà
- 2/5
- Tempo stimato
- 1-3 ore
- Idoneità per principianti
- 78/100
- Tipo di issue
- Bug
- Chiarezza
- Specificata chiaramente
- Stato di attività
- Attiva
- Stack tecnologico
- python
- Ambito
- distributed-systems
Direzione di ricerca
Inizia da Sender.run_once in kafka/producer/sender.py, dove _maybe_abort_batches chiama self._accumulator.abort_batches(exc), poi controlla kafka/producer/record_accumulator.py, che su questi branch definisce solo _abort_batches. Conferma la discrepanza riproducendola con due producer transazionali che condividono lo stesso transactional_id su un broker. È completato quando l'errore fatale si propaga da commit_transaction() invece che il thread I/O registri un AttributeError e continui a ciclare. Prima, ottieni la conferma del maintainer che accetta il backport sui branch 2.2 e 2.3.
Scritto dal modello di indicizzazione a partire dal testo della issue.
Descrizione
On the 2.2 and 2.3 branches (seen on 2.2.15, and still there in 2.2.20, 2.3.2 and the current heads of both), once a transactional producer hits a fatal error the sender thread loops forever, so commit_transaction() never returns.
Sender.run_once calls _maybe_abort_batches(last_error) when the transaction manager has a fatal error, which calls self._accumulator.abort_batches(exc) (sender.py:335 on 2.2, :343 on 2.3), but RecordAccumulator on those branches only has _abort_batches (record_accumulator.py:623 on 2.2, :473 on 2.3). So it raises AttributeError, the I/O thread catches and logs it, and goes round again:
ERROR:kafka.producer.sender:<Sender client_id=kafka-python-producer-2 transactional_id=pytx-producer>: Aborting producer batches due to fatal error: [Error 47] InvalidProducerEpochError
ERROR:kafka.producer.sender:<Sender client_id=kafka-python-producer-2 transactional_id=pytx-producer>: Uncaught error in kafka producer I/O thread
...
AttributeError: 'RecordAccumulator' object has no attribute 'abort_batches'. Did you mean: '_abort_batches'?
It seems master fixed this as part of #3010 (which renamed _abort_batches to abort_batches), so 3.0.x is fine, but the rename never made it back to 2.2 or 2.3.
To reproduce, fence a transactional producer by starting a second one with the same transactional_id, then use the first one again:
from kafka import KafkaProducer
first = KafkaProducer(bootstrap_servers=bootstrap, transactional_id="tx")
first.init_transactions()
second = KafkaProducer(bootstrap_servers=bootstrap, transactional_id="tx")
second.init_transactions()
first.begin_transaction()
first.send("topic", value=b"zombie")
first.commit_transaction() # never returns; the log fills with the AttributeError above
I'd expect commit_transaction() to raise the fatal error (InvalidProducerEpochError here, from AddPartitionsToTxn) instead. I hit this testing a Kafka-protocol gateway against kafka-python 2.2.15, where it's easy to work around, but hopefully a backport of the one-line rename (or a call to _abort_batches in sender.py) to 2.2 and 2.3 is simple enough. I'm happy to open PRs against both branches if that helps.
- Lingua principale
- Python
- Stelle
- 5.9k
- Fork
- 1.5k
- Metriche di merge delle PR
- Nessuna PR unita negli ultimi 30g
Preparare l'ambiente
Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.
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 dpkp/kafka-python
-
StickyPartitionAssignor.assign() loops forever when members subscribe to different topicsForse già presa @mstojcevich l’ha presa 9 giorni fa. Aperta
Difficoltà 4/5 3-5 giorni Idoneità per principianti 15/100
dpkp/kafka-python#3163 ·
-
Difficoltà 1/5 Meno di un'ora Idoneità per principianti 12/100
dpkp/kafka-python#3162 ·
-
Sender: KeyError from muted.remove() while expiring batches orphans the rest; flush() hangs foreverForse già presa @zhechen-cs l’ha presa 8 giorni fa. Aperta
Difficoltà 3/5 1-2 giorni Idoneità per principianti 76/100
dpkp/kafka-python#3159 ·
-
Difficoltà 4/5 3-5 giorni Idoneità per principianti 35/100
dpkp/kafka-python#3122 · 13 commenti ·
-
consumer
Difficoltà 5/5 Più di una settimana Idoneità per principianti 30/100
dpkp/kafka-python#3000 · 2 commenti ·
Tutte le issue di dpkp/kafka-python
Issue simili
-
Difficoltà 2/5 1-3 ore Idoneità per principianti 83/100
I maintainer di solito rispondono entro 1 giorno
-
Difficoltà 2/5 1-3 ore Idoneità per principianti 86/100
google-ai-edge/litert-torch#1346 ·
I maintainer di solito rispondono entro 1 giorno
-
Difficoltà 2/5 1-3 ore Idoneità per principianti 72/100
I maintainer di solito rispondono entro 1 giorno
-
Difficoltà 2/5 1-3 ore Idoneità per principianti 78/100
geneontology/gocam-py#272 ·
-
pushed_authorization_request sends list authorization_details as a Python reprForse già presa @RaphaelFakhri l’ha presa oggi. Aperta
Difficoltà 2/5 1-3 ore Idoneità per principianti 78/100
auth0/auth0-python#903 ·
I maintainer di solito rispondono entro 1 giorno