testcontainers / testcontainers/testcontainers-python
Bug: Segmentation fault when using with confluent kafka consumer
Personne n'a encore pris cette issue.
- Langage dominant
- Python
- Étoiles
- 2.3k
- Forks
- 386
- Merge moyen
- 4 h 40 min
- PR mergées (30 j)
- 1
Description
Describe the bug
Creating an asyncio test case that uses the kafka container on completion of the test (the test itself is successful) the pytest environment segfaults.
To Reproduce
Provide a self-contained code snippet that illustrates the bug or unexpected behavior. Ideally, send a Pull Request to illustrate with a test that illustrates the problem.
import asyncio
import time
import sys
from confluent_kafka import Consumer
import pytest
from testcontainers.core.waiting_utils import wait_for_logs
from testcontainers.kafka import KafkaContainer
@pytest.fixture(scope="module")
def kafka_container():
"""Provide a temporary Redis container."""
with KafkaContainer("mirror.gcr.io/confluentinc/cp-kafka:7.6.0") as container:
wait_for_logs(container, "Awaiting socket connections on 0.0.0.0:9093")
for topic in ["mytopic"]:
container.exec(
f"kafka-topics "
f"--bootstrap-server localhost:9092 "
f"--create "
f"--topic {topic} "
f"--partitions 3 "
f"--replication-factor 1",
)
yield container
async def kafka_consumer(consumer, timeout=1.0):
while True:
msg = await asyncio.to_thread(consumer.poll, timeout=timeout)
if msg is None:
continue
if msg.error():
print("Consumer error: {}".format(msg.error()))
continue
print('Received message: {}'.format(msg.value().decode('utf-8')))
async def main(consumer):
task = asyncio.create_task(kafka_consumer(consumer))
await asyncio.sleep(10)
print("Cancelling the task...")
task.cancel()
try:
await task
except asyncio.CancelledError:
print("Task was cancelled")
@pytest.mark.asyncio(loop_scope="function")
async def test_sample(kafka_container):
consumer = Consumer({
'bootstrap.servers': kafka_container.get_bootstrap_server(),
'group.id': 'mygroup',
'auto.offset.reset': 'earliest',
"debug":"all"})
consumer.subscribe(['mytopic'])
await main(consumer)
consumer.close()
Runtime environment
Provide a summary of your runtime environment. Which operating system, python version, and docker version are you using? What is the version of testcontainers-python you are using? You can run the following commands to get the relevant information.
[project]
name = "kafka-test"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"confluent-kafka>=2.9.0",
]
[dependency-groups]
dev = [
"pytest>=8.3.5",
"pytest-asyncio>=0.26.0",
"testcontainers>=4.10.0",
]
More info
https://github.com/confluentinc/confluent-kafka-python/issues/1953#issuecomment-2784234429
Guide de contribution
Ouvrir le guide de contribution
Par où commencer
- Lisez l'issue en entier, puis le guide de contribution du projet.
- Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
- Forkez le dépôt et travaillez sur une branche.
- Ouvrez une pull request qui référence le numéro de l'issue.
Piste de recherche
Commencez par la reproduction fournie de test_sample en utilisant KafkaContainer, confluent_kafka.Consumer, asyncio.to_thread et pytest-asyncio. Consultez l’issue confluent-kafka liée pour obtenir le contexte associé et consignez les versions de Python, Docker, testcontainers-python et des dépendances. Le travail est considéré comme terminé lorsque le test s’achève et que l’environnement pytest ne segfault plus.
Rédigé par le modèle d'indexation à partir du texte de l'issue.
Évaluation
- Stack technique
- docker, kafka, python
- Domaine
- backend, testing
- Type d'issue
- Bug
- Difficulté
- 4/5
- Temps estimé
- 3-5 jours
- Activité
- À l'abandon
- Clarté
- À clarifier
- Accessibilité débutants
- 35/100