testcontainers / testcontainers/testcontainers-python

Bug: Segmentation fault when using with confluent kafka consumer

未关闭
#805 1 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

还没有人认领这个 Issue。

主要语言
Python
星标
2.3k
派生
386
平均合并
4 小时 40 分钟
30 天内合并 PR
1

描述

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

贡献指南

打开贡献指南

从这里开始

  1. 先读完整个 Issue,再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库,在一个分支上完成修改。
  4. 提交 Pull Request,并在描述里引用这个 Issue 编号。

调研方向

使用 KafkaContainer、confluent_kafka.Consumer、asyncio.to_thread 和 pytest-asyncio,从提供的 test_sample 复现开始。查看链接的 confluent-kafka issue,了解相关背景,并记录 Python、Docker、testcontainers-python 以及依赖项的版本。测试完成且 pytest 环境不再发生 segfault,即表示完成。

由索引模型根据 Issue 内容生成。

评估

技术栈
docker, kafka, python
领域
backend, testing
Issue 类型
缺陷
难度
4/5
预计耗时
3-5 天
活跃度
停滞
描述清晰度
需要澄清
新手友好度
35/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。