DataReader Listener Thread Safety

The following code appears to trigger a deadlock when attempting to use a Listener on a DataReader.

I'm using CycloneDDS commit: d38e63ff8ed5123650beab9bef2b294b56628696: 28 Jun 23
and CycloneDDS Python commit: 7486f5504a988efbd4838b8ecb881d20e34a7644: 25 May 23
and installing the Python library via CYCLONEDDS_HOME=${CYCLONEDDS_HOME} python3 -m pip install /path/to/cyclonedds-python

from concurrent.futures import ThreadPoolExecutor, wait
from dataclasses import dataclass
from time import sleep
from cyclonedds.pub import DataWriter
from cyclonedds.sub import DataReader
from cyclonedds.core import Listener
from cyclonedds.topic import Topic
from cyclonedds.domain import DomainParticipant
from cyclonedds.idl.annotations import key as _key
from cyclonedds.idl import IdlStruct


@dataclass
class KeyedString(IdlStruct, typename="DDS.KeyedString"):
    "Represents a key-value string pair"
    key: str
    value: str
    _key("key")


class MyListener(Listener):
    def on_data_available(self, _):
        print("on_data_available")
        for i in range(10):
            print(i)
            sleep(0.1)


def test_listener():

    participant = DomainParticipant()
    topic = Topic(domain_participant=participant,
                topic_name="foo", data_type=KeyedString)
    writer = DataWriter(publisher_or_participant=participant, topic=topic)
    _ = DataReader(subscriber_or_participant=participant,
                topic=topic, listener=MyListener())

    def do_write():
        writer.write(
            sample=KeyedString("hello", "world"))

    with ThreadPoolExecutor(max_workers=2) as thread_pool:
        wait([thread_pool.submit(do_write) for _ in range(10)])


if __name__ == '__main__':
    test_listener()

Output:

on_data_available
0
(then it just hangs here until I kill the python process)

Setting the DataReader listener to None resolves the deadlock. Is this an issue with the listener (or elsewhere), or am I doing something wrong here?

Start by running the provided reproducer and trace DataReader listener dispatch alongside DataWriter.write and MyListener.on_data_available. Inspect the listener and reader entry points involved in that interaction; done means the example completes without hanging while listener callbacks still run as expected.