Segmentation faults (and the like) and failure to get assignments in multithreaded/asyncio pytest environment

Description
===========
I have a pytest suite that will:

1. Create a kafka testcontainer as a module fixture
2. Create a bunch of topics as a module fixture using the admin client
3. For each test case:
i. Create one Consumer object per topic in a threadpool, all with auto.offset.reset=earliest
ii. Subscribe to them in a threadpool.
iii. Wait for each Consumer to receive an assignment.
iv. In a threadpool, create a bunch of Producers, send messages to the topics, and flush.
v. Wait for each Consumer to process the correct number of messages.
vi. Finally, close all the consumers, regardless of exceptions raised elsewhere.

This test suite is notoriously prone to segmentation faults and the like that crash the entire interpreter and are very disruptive.

I have heard the confluent_kafka is thread safe, but I have not experienced that to be the case. And I'm open to the possibility that this is user error on my part. If so, please, show me the way.

The errors tend to happen after a first test case has run and during the second test case where the Consumer is attempting to monitor for assignments.

There are many different presentations:

```
INTERNAL ERROR: librdkafka rd_kafka_poll_cb:4113: Can't handle op type XMIT_BUF (0x8)
Assertion failed: (!*"INTERNAL ERROR IN LIBRDKAFKA"), function rd_kafka_poll_cb, file rdkafka.c, line 4113.
Fatal Python error: Aborted
```

```
INTERNAL ERROR: librdkafka rd_kafka_poll_cb:4113: Can't handle op type NODE_UPDATE (0x7)
Assertion failed: (!*"INTERNAL ERROR IN LIBRDKAFKA"), function rd_kafka_poll_cb, file rdkafka.c, line 4113.
```

```
INTERNAL ERROR: librdkafka rd_kafka_poll_cb:4113: Can't handle op type CONNECT (0x35)
Assertion failed: (!*"INTERNAL ERROR IN LIBRDKAFKA"), function rd_kafka_poll_cb, file rdkafka.c, line 4113.
```

```
INTERNAL ERROR: librdkafka rd_kafka_poll_cb:4113: Can't handle op type REPLY:GET_REBALANCE_PROTOCOL (0x4000003a)
Assertion failed: (!*"INTERNAL ERROR IN LIBRDKAFKA"), function rd_kafka_poll_cb, file rdkafka.c, line 4113.
```

```
segmentation fault
```

```
INTERNAL ERROR: librdkafka rd_kafka_poll_cb:4113: Can't handle op type REPLY:NODE_UPDATE (0x40000007)
Assertion failed: (!*"INTERNAL ERROR IN LIBRDKAFKA"), function rd_kafka_poll_cb, file rdkafka.c, line 4113.
```

Additionally, if the timeout argument to poll() is set, the consumer never appears to get an assignment at all.

How to reproduce
================

Here's a pretty elaborate script that reliably reproduces the total range of errors that I see with a lot of different knobs to tweak.

```python
import argparse
import asyncio
import logging
import os
import sys
import threading
import time
from asyncio import CancelledError, Task
from contextlib import ExitStack
from functools import partial

import anyio.to_thread
from confluent_kafka import Consumer, KafkaError, KafkaException, Producer
from confluent_kafka.admin import AdminClient, NewTopic

logging.basicConfig(level=logging.INFO)
confluent_kafka_logger = logging.getLogger("confluent_kafka")
confluent_kafka_logger.setLevel(logging.INFO)
confluent_kafka_logger.addHandler(logging.StreamHandler())

CONFIG = {
"NUM_TOPICS": 1,
"NUM_MESSAGES": 1,
"NUM_EXPERIMENTS": sys.maxsize,
"TOPIC_PREFIX": "test-topic",
"MAX_NUM_THREADS": 200,
"KAFKA_BOOTSTRAP_SERVERS": None,
"LOCK_CONSUMER_OPERATIONS": False,
"POLL_TIMEOUT": None,
"CHECK_ASSIGNMENTS_TIMEOUT": 15,
"CHECK_MESSAGES_ARE_CONSUMED_TIMEOUT": 15,
**os.environ,
}

SINGLETONS = {}

def recreate_topic(topic_name):
kafka_config = {
"bootstrap.servers": CONFIG["KAFKA_BOOTSTRAP_SERVERS"],
}
admin_client = AdminClient(kafka_config)

topics = admin_client.list_topics(timeout=10).topics

if topic_name in topics:
print(f"Deleting existing topic '{topic_name}'...")
fs = admin_client.delete_topics([topic_name], operation_timeout=30)
for topic, f in fs.items():
try:
f.result()
print(f"Topic '{topic}' successfully deleted.")
except KafkaException as e:
print(f"Failed to delete topic '{topic}': {e}")

while topic_name in admin_client.list_topics().topics:
time.sleep(1)

print(f"Confirmed that topic {topic_name} does not exist.")

print(f"Creating topic '{topic_name}'...")
new_topic = NewTopic(topic_name, num_partitions=3, replication_factor=1)
try:
admin_client.create_topics([new_topic], operation_timeout=30)
print(f"Topic '{topic_name}' created.")
except KafkaException as e:
print(f"Failed to create topic '{topic_name}': {e}")

# Loop until topic exists
while topic_name not in admin_client.list_topics().topics:
time.sleep(1)

print(f"Confirmed that topic {topic_name} exists.")

class LockingConsumer:
"""A patched version of the Consumer class that locks operations."""

def __init__(self, config: dict, logger):
self.config = config
self.consumer = Consumer(config, logger=logger)
self.lock = threading.Lock()

def __getattr__(self, item):
obj = getattr(self.consumer, item)
if callable(obj):
return self._wrap_callable(obj)

def _wrap_callable(self, func):

def wrapped(*args, **kwargs):
with self.lock:
return func(*args, **kwargs)

return wrapped

class AsyncConsumer:

def __init__(self, topic_name, num_messages):
self.topic_name = topic_name
self.num_messages = num_messages
consumer_config = {
"bootstrap.servers": CONFIG["KAFKA_BOOTSTRAP_SERVERS"],
"group.id": f"test_group.{topic_name}",
"auto.offset.reset": "latest",
"enable.auto.commit": "false",
}
if CONFIG["LOCK_CONSUMER_OPERATIONS"]:
consumer_class = LockingConsumer
else:
consumer_class = Consumer

self.consumer = consumer_class(consumer_config, logger=confluent_kafka_logger)

self.task: Task | None = None

def _log(self, msg):
print(f"Consumer({self.topic_name}): {msg}.")

async def assignments(self):
return await anyio.to_thread.run_sync(
self.consumer.assignment, limiter=SINGLETONS["limiter"]
)

async def start(self):
self._log("Subscribing.")
await anyio.to_thread.run_sync(
self.consumer.subscribe, [self.topic_name], limiter=SINGLETONS["limiter"]
)
try:
self.task = asyncio.create_task(self.poll_loop())
except CancelledError:
self._log("Timeout.")
except KafkaException as e:
self._log(f"{e}")

async def close(self):
if self.task is not None and not self.task.done():
self._log("Cancelling poll_loop().")
self.task.cancel()
else:
self._log("poll_loop() already done.")
self._log("Closing consumer")
await anyio.to_thread.run_sync(
self.consumer.close, limiter=SINGLETONS["limiter"]
)
self._log("Consumer closed")

async def poll_loop(self):
while True:
self._log("Polling.")
msg = await anyio.to_thread.run_sync(
(
self.consumer.poll
if CONFIG["POLL_TIMEOUT"] is None
else partial(self.consumer.poll, timeout=CONFIG["POLL_TIMEOUT"])
),
limiter=SINGLETONS["limiter"],
)
self._log(f"Message received: {msg.value()}")
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
raise KafkaException(msg.error())
await anyio.to_thread.run_sync(
self.consumer.commit, limiter=SINGLETONS["limiter"]
)
self.num_messages -= 1
self._log(f"{self.num_messages} messages left.")

async def add_messages(topic_name):
producer_config = {"bootstrap.servers": CONFIG["KAFKA_BOOTSTRAP_SERVERS"]}
producer = Producer(producer_config)
for i in range(CONFIG["NUM_MESSAGES"]):
print(f"Producer({topic_name}) - producing message {i}.")
await anyio.to_thread.run_sync(
producer.produce, topic_name, f"{i}", limiter=SINGLETONS["limiter"]
)
await anyio.to_thread.run_sync(producer.poll, 0, limiter=SINGLETONS["limiter"])
await anyio.to_thread.run_sync(producer.flush, limiter=SINGLETONS["limiter"])

async def check_assignments(consumer_handlers):
assignments = []
while not assignments:
handler_assignments = await asyncio.gather(
*[consumer_handler.assignments() for consumer_handler in consumer_handlers]
)
is_assigned = [len(assignment) > 0 for assignment in handler_assignments]
print(f"Assignments: {is_assigned}")
if sum(is_assigned) != len(consumer_handlers):
print(f"Only {sum(is_assigned)} handlers have assignments, sleeping")
await asyncio.sleep(1)
else:
print("All handlers have assignments.")
return

async def check_messages_are_consumed(consumer_handlers):
while any(
consumer_handler.num_messages > 0 for consumer_handler in consumer_handlers
):
await asyncio.sleep(0.1)

async def run_test_case(topic_names):
print("--- Starting consumers. ---")
consumer_handlers = [
AsyncConsumer(topic_name, CONFIG["NUM_MESSAGES"]) for topic_name in topic_names
]
await asyncio.gather(
*[consumer_handler.start() for consumer_handler in consumer_handlers]
)

try:
print("--- Waiting for assignments. ---")
await asyncio.wait_for(
check_assignments(consumer_handlers),
timeout=CONFIG["CHECK_ASSIGNMENTS_TIMEOUT"],
)

print("--- Adding messages. ---")
await asyncio.gather(*[add_messages(topic_name) for topic_name in topic_names])
print("--- Waiting for messages to be consumed. ---")
await asyncio.wait_for(
check_messages_are_consumed(consumer_handlers),
timeout=CONFIG["CHECK_MESSAGES_ARE_CONSUMED_TIMEOUT"],
)
finally:
print("--- Closing consumers. ---")
await asyncio.gather(
*[consumer_handler.close() for consumer_handler in consumer_handlers]
)
print("--- Test case done. ---")

async def run_experiments(topic_names):
for i in range(CONFIG["NUM_EXPERIMENTS"]):
print(f"------ Running experiment {i} ------")
await run_test_case(topic_names)

async def create_topics():
tasks = []
topic_names = []
for i in range(CONFIG["NUM_TOPICS"]):
topic_name = f"{CONFIG['TOPIC_PREFIX']}-{i}"
topic_names.append(topic_name)
tasks.append(
anyio.to_thread.run_sync(
recreate_topic, topic_name, limiter=SINGLETONS["limiter"]
)
)
await asyncio.gather(*tasks)
return topic_names

async def run_test_application():
topic_names = await create_topics()
await run_experiments(topic_names)

if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--num-topics", type=int, default=CONFIG["NUM_TOPICS"])
parser.add_argument("--num-messages", type=int, default=CONFIG["NUM_MESSAGES"])
parser.add_argument(
"--num-experiments", type=int, default=CONFIG["NUM_EXPERIMENTS"]
)
parser.add_argument("--topic-prefix", type=str, default=CONFIG["TOPIC_PREFIX"])
parser.add_argument(
"--max-num-threads", type=int, default=CONFIG["MAX_NUM_THREADS"]
)
parser.add_argument("--lock-consumer-operations", action="store_true")
parser.add_argument("--poll-timeout", type=float, default=CONFIG["POLL_TIMEOUT"])
parser.add_argument(
"--check-assignments-timeout",
type=float,
default=CONFIG["CHECK_ASSIGNMENTS_TIMEOUT"],
)
parser.add_argument(
"--check-messages-are-consumed-timeout",
type=float,
default=CONFIG["CHECK_MESSAGES_ARE_CONSUMED_TIMEOUT"],
)
parser.add_argument(
"--kafka-bootstrap-servers", type=str, default=CONFIG["KAFKA_BOOTSTRAP_SERVERS"]
)
args = parser.parse_args()
CONFIG["NUM_TOPICS"] = args.num_topics
CONFIG["NUM_MESSAGES"] = args.num_messages
CONFIG["NUM_EXPERIMENTS"] = args.num_experiments
CONFIG["TOPIC_PREFIX"] = args.topic_prefix
CONFIG["KAFKA_BOOTSTRAP_SERVERS"] = args.kafka_bootstrap_servers
CONFIG["MAX_NUM_THREADS"] = args.max_num_threads
CONFIG["LOCK_CONSUMER_OPERATIONS"] = args.lock_consumer_operations
CONFIG["POLL_TIMEOUT"] = args.poll_timeout
CONFIG["CHECK_ASSIGNMENTS_TIMEOUT"] = args.check_assignments_timeout
CONFIG["CHECK_MESSAGES_ARE_CONSUMED_TIMEOUT"] = (
args.check_messages_are_consumed_timeout
)

with ExitStack() as stack:
if CONFIG["KAFKA_BOOTSTRAP_SERVERS"] is None:
import testcontainers.kafka

kafka = stack.enter_context(testcontainers.kafka.KafkaContainer())
CONFIG["KAFKA_BOOTSTRAP_SERVERS"] = kafka.get_bootstrap_server()

print(f"NUM_TOPICS: {CONFIG['NUM_TOPICS']}")
print(f"NUM_MESSAGES: {CONFIG['NUM_MESSAGES']}")
print(f"NUM_EXPERIMENTS: {CONFIG['NUM_EXPERIMENTS']}")
print(f"TOPIC_PREFIX: {CONFIG['TOPIC_PREFIX']}")
print(f"KAFKA_BOOTSTRAP_SERVERS: {CONFIG['KAFKA_BOOTSTRAP_SERVERS']}")
print(f"MAX_NUM_THREADS: {CONFIG['MAX_NUM_THREADS']}")
print(f"LOCK_CONSUMER_OPERATIONS: {CONFIG['LOCK_CONSUMER_OPERATIONS']}")
print(f"POLL_TIMEOUT: {CONFIG['POLL_TIMEOUT']}")
print(f"CHECK_ASSIGNMENTS_TIMEOUT: {CONFIG['CHECK_ASSIGNMENTS_TIMEOUT']}")
print(
f"CHECK_MESSAGES_ARE_CONSUMED_TIMEOUT: {CONFIG['CHECK_MESSAGES_ARE_CONSUMED_TIMEOUT']}"
)

# This breaks wait_for_assignments() because the first call to poll() causes
# a lock issue
if CONFIG["LOCK_CONSUMER_OPERATIONS"] and CONFIG["POLL_TIMEOUT"] in (None, -1):
print(
"Warning, blocking poll() call will prevent another thread from acquiring the lock."
)

# For whatever reason, consumers don't get assigned partitions if this is
# set to a value other than None. Even passing in None to poll() causes issues
if CONFIG["POLL_TIMEOUT"] not in (None, -1):
print(
"Warning, setting POLL_TIMEOUT typically prevents assignments from ever happening."
)

SINGLETONS["limiter"] = anyio.CapacityLimiter(CONFIG["MAX_NUM_THREADS"])

asyncio.run(run_test_application(), debug=True)

```

### Assignments never happening when poll(1.0)

https://gist.github.com/andreaimprovised/6221cba7c0be98ee3189dd517998bda3

### INTERNAL ERROR with 3 topics

https://gist.github.com/andreaimprovised/d80eedeea6ef7beb44fff228df1942da

### segmentation fault with 12 topics

https://gist.github.com/andreaimprovised/5bc6acdc05fecb35d7cb7f20295c31f7

### Segmentation fault with just 1 topic and 1 message per test case

https://gist.github.com/andreaimprovised/1fe7b9f8be0d34d8a6dc40827c802934

Additional requirements:

```
testcontainers==4.7.2
anyio==4.4.0
```

I'm currently using python 3.10.

Checklist
=========
Please provide the following information:

- [x] confluent-kafka-python and librdkafka version (`confluent_kafka.version()` and `confluent_kafka.libversion()`):

In [2]: confluent_kafka.version()
Out[2]: ('2.5.0', 33882112)

- [x] Apache Kafka broker version:

confluentinc/cp-kafka:7.6.0

- [x] Client configuration: `{...}`

It's in the code.

- [x] Operating system:

I've seen this on darwin arm64 and linux x86_64.

- [x] Provide client logs (with `'debug': '..'` as necessary)

Here is an example

- [ ] Provide broker log excerpts

Hmmm, I'll try to figure out how to get these.

- [ ] Critical issue

It's not critical, but it depends on how critical you think automated test suites are.