Transactional producer's sender loops on AttributeError after a fatal error (2.2 and 2.3 branches)
Nadie ha tomado este issue todavía.
Evaluación
- Dificultad
- 2/5
- Tiempo estimado
- 1-3 horas
- Aptitud para principiantes
- 78/100
- Tipo de issue
- Error
- Claridad
- Bien especificado
- Estado de actividad
- Activo
- Stack tecnológico
- python
- Área
- distributed-systems
Línea de trabajo
Empieza en Sender.run_once en kafka/producer/sender.py, donde _maybe_abort_batches llama a self._accumulator.abort_batches(exc), y luego revisa kafka/producer/record_accumulator.py, que solo define _abort_batches en estas ramas. Confirma la discrepancia reproduciéndola con dos productores transaccionales que comparten un mismo transactional_id contra un broker. Está terminado cuando el error fatal se propaga fuera de commit_transaction() en lugar de que el hilo de I/O registre un AttributeError y siga en bucle. Antes, confirma que el mantenedor acepta el backport en las ramas 2.2 y 2.3.
Escrito por el modelo de indexación a partir del texto del issue.
Descripción
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.
- Lenguaje dominante
- Python
- Estrellas
- 5.9k
- Forks
- 1.5k
- Métricas de merge de PR
- Sin PR fusionados en 30 d
Preparar el entorno
Este proyecto no incluye contenedor de desarrollo, Dockerfile ni guía de contribución, así que la configuración corre por tu cuenta: empieza por su README y consulta nuestra guía para la primera contribución para los pasos generales.
Primeros pasos
- Lee el issue completo y luego la guía de contribución del proyecto.
- Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
- Haz un fork del repositorio y trabaja en una rama.
- Abre un pull request que haga referencia al número del issue.
Más de dpkp/kafka-python
-
StickyPartitionAssignor.assign() loops forever when members subscribe to different topicsPosiblemente ocupada @mstojcevich la tomó hace 9 días. Abierto
Dificultad 4/5 3-5 días Aptitud para principiantes 15/100
dpkp/kafka-python#3163 ·
-
Dificultad 1/5 Menos de una hora Aptitud para principiantes 12/100
dpkp/kafka-python#3162 ·
-
Sender: KeyError from muted.remove() while expiring batches orphans the rest; flush() hangs foreverPosiblemente ocupada @zhechen-cs la tomó hace 8 días. Abierto
Dificultad 3/5 1-2 días Aptitud para principiantes 76/100
dpkp/kafka-python#3159 ·
-
Dificultad 4/5 3-5 días Aptitud para principiantes 35/100
dpkp/kafka-python#3122 · 13 comentarios ·
-
consumer
Dificultad 5/5 Más de una semana Aptitud para principiantes 30/100
dpkp/kafka-python#3000 · 2 comentarios ·
Todos los issues de dpkp/kafka-python
Issues similares
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 83/100
Los mantenedores suelen responder en 1 día
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 86/100
google-ai-edge/litert-torch#1346 ·
Los mantenedores suelen responder en 1 día
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 72/100
Los mantenedores suelen responder en 1 día
-
Dificultad 2/5 1-3 horas Aptitud para principiantes 78/100
geneontology/gocam-py#272 ·
-
pushed_authorization_request sends list authorization_details as a Python reprPosiblemente ocupada @RaphaelFakhri la tomó hoy. Abierto
Dificultad 2/5 1-3 horas Aptitud para principiantes 78/100
auth0/auth0-python#903 ·
Los mantenedores suelen responder en 1 día